File System Port

file_system_port

File system port for abstract file operations.

Defines the FileSystemPort contract that application code depends on,
decoupling file I/O from any specific implementation.

FileSystemPort

Bases: OutboundPort

Abstract contract for file system operations.

Responsibilities

• Read and write file contents as bytes.
• Delete files and check existence.
• List directory contents.

Non-Responsibilities

• Manage file permissions or ownership.
• Resolve symlinks or handle special file types.
• Provide atomic or transactional file operations.
• Stream large files — the whole content is read into memory.

Example

fs = MyFileSystem()
await fs.write("data.json", b'{"key": "value"}')
content = await fs.read("data.json")

Source code in src/forging_blocks/application/ports/outbound/file_system_port.py

class FileSystemPort(
    OutboundPort,
):
    """Abstract contract for file system operations.

    Responsibilities:
        - Read and write file contents as bytes.
        - Delete files and check existence.
        - List directory contents.

    Non-Responsibilities:
        - Manage file permissions or ownership.
        - Resolve symlinks or handle special file types.
        - Provide atomic or transactional file operations.
        - Stream large files — the whole content is read into memory.

    Example:
        ```python
        fs = MyFileSystem()
        await fs.write("data.json", b'{"key": "value"}')
        content = await fs.read("data.json")
        ```
    """

    @abstractmethod
    async def read(self, path: Path | str) -> bytes:
        """Read the contents of a file.

        Args:
            path: Path to the file.

        Returns:
            The file contents as bytes.

        Raises:
            FileNotFoundError: If the file does not exist.

        """
        ...

    @abstractmethod
    async def write(self, path: Path | str, data: bytes) -> None:
        """Write data to a file, creating parent directories as needed.

        Args:
            path: Path to the file.
            data: The bytes to write.

        """

    @abstractmethod
    async def delete(self, path: Path | str) -> None:
        """Delete a file.

        Args:
            path: Path to the file.

        Raises:
            FileNotFoundError: If the file does not exist.

        """

    @abstractmethod
    async def exists(self, path: Path | str) -> bool:
        """Check whether a file or directory exists.

        Args:
            path: Path to check.

        Returns:
            ``True`` if the path exists, ``False`` otherwise.

        """
        ...

    @abstractmethod
    async def list_dir(self, path: Path | str) -> list[Path]:
        """List the contents of a directory.

        Args:
            path: Path to the directory.

        Returns:
            A list of ``Path`` entries in the directory.

        Raises:
            NotADirectoryError: If *path* is not a directory.

        """
        ...

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

Read the contents of a file.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the file.	required

Returns:

Type	Description
bytes	The file contents as bytes.

Raises:

Type	Description
FileNotFoundError	If the file does not exist.

Source code in src/forging_blocks/application/ports/outbound/file_system_port.py

@abstractmethod
async def read(self, path: Path | str) -> bytes:
    """Read the contents of a file.

    Args:
        path: Path to the file.

    Returns:
        The file contents as bytes.

    Raises:
        FileNotFoundError: If the file does not exist.

    """
    ...

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

Write data to a file, creating parent directories as needed.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the file.	required
data	bytes	The bytes to write.	required

Source code in src/forging_blocks/application/ports/outbound/file_system_port.py

@abstractmethod
async def write(self, path: Path | str, data: bytes) -> None:
    """Write data to a file, creating parent directories as needed.

    Args:
        path: Path to the file.
        data: The bytes to write.

    """

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

Delete a file.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the file.	required

Raises:

Type	Description
FileNotFoundError	If the file does not exist.

Source code in src/forging_blocks/application/ports/outbound/file_system_port.py

@abstractmethod
async def delete(self, path: Path | str) -> None:
    """Delete a file.

    Args:
        path: Path to the file.

    Raises:
        FileNotFoundError: If the file does not exist.

    """

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

Check whether a file or directory exists.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to check.	required

Returns:

Type	Description
bool	True if the path exists, False otherwise.

Source code in src/forging_blocks/application/ports/outbound/file_system_port.py

@abstractmethod
async def exists(self, path: Path | str) -> bool:
    """Check whether a file or directory exists.

    Args:
        path: Path to check.

    Returns:
        ``True`` if the path exists, ``False`` otherwise.

    """
    ...

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

List the contents of a directory.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the directory.	required

Returns:

Type	Description
list[Path]	A list of Path entries in the directory.

Raises:

Type	Description
NotADirectoryError	If path is not a directory.

Source code in src/forging_blocks/application/ports/outbound/file_system_port.py

@abstractmethod
async def list_dir(self, path: Path | str) -> list[Path]:
    """List the contents of a directory.

    Args:
        path: Path to the directory.

    Returns:
        A list of ``Path`` entries in the directory.

    Raises:
        NotADirectoryError: If *path* is not a directory.

    """
    ...