Class swarmauri_storage_file.file_storage_adapter.FileStorageAdapter

swarmauri_storage_file.file_storage_adapter.FileStorageAdapter

FileStorageAdapter(output_dir, *, prefix='', **kwargs)

Bases: StorageAdapterBase

Write and read artefacts on the local disk.

Source code in swarmauri_storage_file/file_storage_adapter.py

def __init__(self, output_dir: str | os.PathLike, *, prefix: str = "", **kwargs):
    super().__init__(**kwargs)
    self._root = Path(output_dir).expanduser().resolve()
    self._root.mkdir(parents=True, exist_ok=True)
    self._prefix = prefix.lstrip("/")

root_uri property

root_uri

Return the workspace root as a file:// URI.

upload

upload(key, data)

Copy data to ${root_dir}/${key} atomically and return the artifact URI.

Source code in swarmauri_storage_file/file_storage_adapter.py

def upload(self, key: str, data: BinaryIO) -> str:
    """Copy *data* to ``${root_dir}/${key}`` atomically and return the artifact URI."""
    dest = self._full_key(key)
    dest.parent.mkdir(parents=True, exist_ok=True)

    tmp = dest.with_suffix(dest.suffix + ".tmp")
    with open(tmp, "wb") as fh:
        shutil.copyfileobj(data, fh)

    tmp.replace(dest)

    return f"{self.root_uri}{key.lstrip('/')}"

download

download(key)

Return a :class:BytesIO with the contents of ${root_dir}/${key}.

Source code in swarmauri_storage_file/file_storage_adapter.py

def download(self, key: str) -> BinaryIO:
    """Return a :class:`BytesIO` with the contents of ``${root_dir}/${key}``."""
    path = self._full_key(key)
    if not path.exists():
        raise FileNotFoundError(path)

    buffer = io.BytesIO(path.read_bytes())
    buffer.seek(0)
    return buffer

upload_dir

upload_dir(src, *, prefix='')

Recursively upload files from src under prefix.

Source code in swarmauri_storage_file/file_storage_adapter.py

def upload_dir(self, src: str | os.PathLike, *, prefix: str = "") -> None:
    """Recursively upload files from *src* under ``prefix``."""
    base = Path(src)
    for path in base.rglob("*"):
        if path.is_file():
            rel = path.relative_to(base)
            key = os.path.join(prefix, rel.as_posix())
            with path.open("rb") as fh:
                self.upload(key, fh)

iter_prefix

iter_prefix(prefix)

Yield stored keys beginning with prefix.

Source code in swarmauri_storage_file/file_storage_adapter.py

def iter_prefix(self, prefix: str):
    """Yield stored keys beginning with ``prefix``."""
    base = self._full_key(prefix)
    if not base.exists():
        return
    for path in base.rglob("*"):
        if path.is_file():
            rel = path.relative_to(self._root)
            yield str(rel)

download_dir

download_dir(prefix, dest_dir)

Copy all files under prefix into dest_dir.

Source code in swarmauri_storage_file/file_storage_adapter.py

def download_dir(self, prefix: str, dest_dir: str | os.PathLike) -> None:
    """Copy all files under ``prefix`` into ``dest_dir``."""
    src_root = self._full_key(prefix)
    dest = Path(dest_dir)
    if not src_root.exists():
        return
    for path in src_root.rglob("*"):
        if path.is_file():
            rel = path.relative_to(src_root)
            target = dest / rel
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(path, target)

from_uri classmethod

from_uri(uri)

Instantiate the adapter from a file:// URI.

Source code in swarmauri_storage_file/file_storage_adapter.py

@classmethod
def from_uri(cls, uri: str) -> "FileStorageAdapter":
    """Instantiate the adapter from a ``file://`` URI."""
    path = Path(uri[7:]).resolve()
    return cls(output_dir=path)