Skip to content

Os File System

os_file_system

Operating-system file system implementation of the FileSystemPort port.

Uses Python 3.14+ asyncio.to_thread() to wrap blocking pathlib
and os calls, avoiding external dependencies like aiofiles.

OSFileSystem

Bases: FileSystemPort

File system implementation backed by pathlib + asyncio.to_thread.

Source code in src/forging_blocks/infrastructure/file_system/os_file_system.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class OSFileSystem(FileSystemPort):
    """File system implementation backed by ``pathlib`` + ``asyncio.to_thread``."""

    async def read(self, path: Path | str) -> bytes:
        """Read file contents as bytes."""
        target = Path(path)
        return await asyncio.to_thread(target.read_bytes)

    async def write(self, path: Path | str, data: bytes) -> None:
        """Write bytes to a file, creating parent directories."""
        target = Path(path)
        target.parent.mkdir(parents=True, exist_ok=True)
        await asyncio.to_thread(target.write_bytes, data)

    async def delete(self, path: Path | str) -> None:
        """Delete a file."""
        target = Path(path)
        await asyncio.to_thread(target.unlink)

    async def exists(self, path: Path | str) -> bool:
        """Check if a path exists."""
        target = Path(path)
        return await asyncio.to_thread(target.exists)

    async def list_dir(self, path: Path | str) -> list[Path]:
        """List directory contents."""
        target = Path(path)
        return await asyncio.to_thread(lambda: list(target.iterdir()))

read(path: Path | str) -> bytes async

Read file contents as bytes.

Source code in src/forging_blocks/infrastructure/file_system/os_file_system.py 
16
17
18
19
async def read(self, path: Path | str) -> bytes:
    """Read file contents as bytes."""
    target = Path(path)
    return await asyncio.to_thread(target.read_bytes)

write(path: Path | str, data: bytes) -> None async

Write bytes to a file, creating parent directories.

Source code in src/forging_blocks/infrastructure/file_system/os_file_system.py 
21
22
23
24
25
async def write(self, path: Path | str, data: bytes) -> None:
    """Write bytes to a file, creating parent directories."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    await asyncio.to_thread(target.write_bytes, data)

delete(path: Path | str) -> None async

Delete a file.

Source code in src/forging_blocks/infrastructure/file_system/os_file_system.py 
27
28
29
30
async def delete(self, path: Path | str) -> None:
    """Delete a file."""
    target = Path(path)
    await asyncio.to_thread(target.unlink)

exists(path: Path | str) -> bool async

Check if a path exists.

Source code in src/forging_blocks/infrastructure/file_system/os_file_system.py 
32
33
34
35
async def exists(self, path: Path | str) -> bool:
    """Check if a path exists."""
    target = Path(path)
    return await asyncio.to_thread(target.exists)

list_dir(path: Path | str) -> list[Path] async

List directory contents.

Source code in src/forging_blocks/infrastructure/file_system/os_file_system.py 
37
38
39
40
async def list_dir(self, path: Path | str) -> list[Path]:
    """List directory contents."""
    target = Path(path)
    return await asyncio.to_thread(lambda: list(target.iterdir()))