Skip to content

File System Port

file_system_port

File system port for abstract file operations.

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

FileSystemPort

Bases: OutboundPort[object, object], Protocol

Structural protocol for file system operations.

Any object with read, write, delete, exists,
and list_dir async methods satisfies this protocol.

Source code in src/forging_blocks/application/ports/outbound/file_system_port.py
class FileSystemPort(OutboundPort[object, object], Protocol):
    """Structural protocol for file system operations.

    Any object with ``read``, ``write``, ``delete``, ``exists``,
    and ``list_dir`` async methods satisfies this protocol.
    """

    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.
        """
        ...

    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.
        """
        ...

    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.
        """
        ...

    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.
        """
        ...

    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 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
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 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
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 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
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 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
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] 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
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.
    """
    ...