forge.storage

Multi-backend file storage abstraction supporting local filesystem, S3, GCS, and Azure.

forge.storage

Classes

AzureAdapter

Storage adapter backed by Azure Blob Storage.

Uses the azure-storage-blob library.

Source code in src/forge/storage/adapters/azure.py

class AzureAdapter:
    """
    Storage adapter backed by Azure Blob Storage.

    Uses the *azure-storage-blob* library.
    """

    def __init__(
        self,
        container: str,
        *,
        connection_string: str | None = None,
        account_url: str | None = None,
        credential: str | None = None,
    ) -> None:
        from azure.storage.blob import BlobServiceClient

        if connection_string:
            self._service = BlobServiceClient.from_connection_string(connection_string)
        elif account_url:
            self._service = BlobServiceClient(account_url=account_url, credential=credential)
        else:
            raise ValueError(
                "Either connection_string or account_url must be provided for AzureAdapter"
            )

        self._container_name = container
        self._container_client = self._service.get_container_client(container)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _key(self, remote_path: str) -> str:
        return normalize_path(remote_path)

    def _file_info_from_blob(self, blob: Any) -> FileInfo:
        last_mod = blob.last_modified
        return FileInfo(
            path=blob.name,
            size=blob.size or 0,
            last_modified=last_mod.replace(tzinfo=UTC) if last_mod else None,
            content_type=blob.content_settings.content_type if blob.content_settings else None,
            etag=blob.etag.strip('"') if blob.etag else None,
            metadata=dict(blob.metadata) if blob.metadata else {},
        )

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def upload(
        self,
        local_path: str,
        remote_path: str,
        *,
        content_type: str | None = None,
        metadata: dict[str, str] | None = None,
        multipart_threshold: int = 100 * 1024 * 1024,
    ) -> FileInfo:
        from azure.storage.blob import ContentSettings

        key = self._key(remote_path)
        ct = content_type or detect_content_type(local_path)

        blob_client = self._container_client.get_blob_client(key)
        content_settings = ContentSettings(content_type=ct)

        file_size = Path(local_path).stat().st_size

        try:
            with Path(local_path).open("rb") as data:
                if file_size > multipart_threshold:
                    blob_client.upload_blob(
                        data,
                        overwrite=True,
                        content_settings=content_settings,
                        metadata=metadata,
                        max_concurrency=4,
                    )
                else:
                    blob_client.upload_blob(
                        data,
                        overwrite=True,
                        content_settings=content_settings,
                        metadata=metadata,
                    )
        except Exception as exc:
            raise StorageUploadError(
                f"Failed to upload {local_path} to azure://{self._container_name}/{key}: {exc}"
            ) from exc

        props = blob_client.get_blob_properties()
        return self._file_info_from_blob(props)

    async def download(
        self,
        remote_path: str,
        local_path: str,
        *,
        overwrite: bool = False,
    ) -> str:
        key = self._key(remote_path)
        dst = Path(local_path)

        if dst.exists() and not overwrite:
            raise StorageDownloadError(
                f"Local file already exists: {local_path} (use overwrite=True to replace)"
            )

        dst.parent.mkdir(parents=True, exist_ok=True)

        blob_client = self._container_client.get_blob_client(key)

        try:
            with dst.open("wb") as data:
                stream = blob_client.download_blob()
                stream.readinto(data)
        except Exception as exc:
            raise StorageDownloadError(
                f"Failed to download azure://{self._container_name}/{key} to {local_path}: {exc}"
            ) from exc

        return str(dst.resolve())

    async def delete(
        self,
        remote_path: str,
    ) -> bool:
        key = self._key(remote_path)
        blob_client = self._container_client.get_blob_client(key)

        if not await self.exists(remote_path):
            return False
        try:
            blob_client.delete_blob()
            return True
        except Exception as exc:
            raise StorageDeleteError(
                f"Failed to delete azure://{self._container_name}/{key}: {exc}"
            ) from exc

    async def list(
        self,
        prefix: str = "",
        *,
        recursive: bool = True,
    ) -> list[FileInfo]:
        key_prefix = self._key(prefix)
        items: list[FileInfo] = []

        try:
            blobs = self._container_client.list_blobs(name_starts_with=key_prefix)
            for blob in blobs:
                name: str = blob.name
                if not recursive and "/" in name[len(key_prefix) :].lstrip("/"):
                    continue
                items.append(self._file_info_from_blob(blob))
        except Exception as exc:
            raise StorageListError(
                f"Failed to list objects at prefix {prefix} in container {self._container_name}: {exc}"
            ) from exc

        items.sort(key=lambda f: f.path)
        return items

    async def exists(
        self,
        remote_path: str,
    ) -> bool:
        key = self._key(remote_path)
        blob_client = self._container_client.get_blob_client(key)
        try:
            blob_client.get_blob_properties()
            return True
        except Exception as exc:
            error_code = getattr(exc, "error_code", None)
            if error_code == "BlobNotFound":
                return False
            raise StorageConnectionError(
                f"Failed to check existence of azure://{self._container_name}/{key}: {exc}"
            ) from exc

    async def presigned_url(
        self,
        remote_path: str,
        *,
        expiration: int = 3600,
    ) -> str:
        from azure.storage.blob import BlobSasPermissions, generate_blob_sas

        key = self._key(remote_path)
        blob_client = self._container_client.get_blob_client(key)

        if not await self.exists(remote_path):
            raise StorageNotFoundError(
                f"Object does not exist: azure://{self._container_name}/{key}"
            )

        account_name = self._service.account_name
        if not account_name:
            raise StorageConnectionError("Azure account name is not set")

        try:
            user_delegation_key = None
            sas_token = generate_blob_sas(
                account_name=account_name,
                container_name=self._container_name,
                blob_name=key,
                account_key=self._service.credential.account_key
                if hasattr(self._service.credential, "account_key")
                else None,
                user_delegation_key=user_delegation_key,
                permission=BlobSasPermissions(read=True),
                expiry=datetime.now(UTC) + timedelta(seconds=expiration),
            )
            url = f"{blob_client.url}?{sas_token}"
            return url
        except Exception as exc:
            raise StorageConnectionError(
                f"Failed to generate SAS URL for azure://{self._container_name}/{key}: {exc}"
            ) from exc

FileInfo dataclass

Metadata for a single object returned by list operations.

Source code in src/forge/storage/adapters/base.py

@dataclass(frozen=True)
class FileInfo:
    """Metadata for a single object returned by list operations."""

    path: str
    size: int = 0
    last_modified: datetime | None = None
    content_type: str | None = None
    etag: str | None = None
    metadata: dict[str, str] = field(default_factory=dict)

GCSAdapter

Storage adapter backed by Google Cloud Storage.

Uses the google-cloud-storage library.

Source code in src/forge/storage/adapters/gcs.py

class GCSAdapter:
    """
    Storage adapter backed by Google Cloud Storage.

    Uses the *google-cloud-storage* library.
    """

    def __init__(
        self,
        bucket: str,
        *,
        project: str | None = None,
        credentials_path: str | None = None,
    ) -> None:
        from google.cloud import storage

        if credentials_path:
            self._client = storage.Client.from_service_account_json(
                credentials_path, project=project
            )
        else:
            self._client = storage.Client(project=project)

        self._bucket = self._client.bucket(bucket)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _key(self, remote_path: str) -> str:
        return normalize_path(remote_path)

    def _file_info_from_blob(self, blob: Any) -> FileInfo:
        last_mod = blob.updated
        return FileInfo(
            path=blob.name,
            size=blob.size or 0,
            last_modified=last_mod.replace(tzinfo=UTC) if last_mod else None,
            content_type=blob.content_type or "application/octet-stream",
            etag=blob.etag.strip('"') if blob.etag else None,
            metadata=dict(blob.metadata) if blob.metadata else {},
        )

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def upload(
        self,
        local_path: str,
        remote_path: str,
        *,
        content_type: str | None = None,
        metadata: dict[str, str] | None = None,
        multipart_threshold: int = 100 * 1024 * 1024,
    ) -> FileInfo:
        key = self._key(remote_path)
        ct = content_type or detect_content_type(local_path)

        blob = self._bucket.blob(key)
        if metadata:
            blob.metadata = metadata

        try:
            blob.upload_from_filename(
                local_path,
                content_type=ct,
                if_generation_match=None,
            )
        except Exception as exc:
            raise StorageUploadError(
                f"Failed to upload {local_path} to gs://{self._bucket.name}/{key}: {exc}"
            ) from exc

        blob.reload()
        return self._file_info_from_blob(blob)

    async def download(
        self,
        remote_path: str,
        local_path: str,
        *,
        overwrite: bool = False,
    ) -> str:
        key = self._key(remote_path)
        dst = Path(local_path)

        if dst.exists() and not overwrite:
            raise StorageDownloadError(
                f"Local file already exists: {local_path} (use overwrite=True to replace)"
            )

        dst.parent.mkdir(parents=True, exist_ok=True)

        blob = self._bucket.blob(key)
        if not blob.exists():
            raise StorageNotFoundError(f"Object does not exist: gs://{self._bucket.name}/{key}")

        try:
            blob.download_to_filename(local_path)
        except Exception as exc:
            raise StorageDownloadError(
                f"Failed to download gs://{self._bucket.name}/{key} to {local_path}: {exc}"
            ) from exc

        return str(dst.resolve())

    async def delete(
        self,
        remote_path: str,
    ) -> bool:
        key = self._key(remote_path)
        blob = self._bucket.blob(key)
        if not blob.exists():
            return False
        try:
            blob.delete()
            return True
        except Exception as exc:
            raise StorageDeleteError(
                f"Failed to delete gs://{self._bucket.name}/{key}: {exc}"
            ) from exc

    async def list(
        self,
        prefix: str = "",
        *,
        recursive: bool = True,
    ) -> list[FileInfo]:
        key_prefix = self._key(prefix)
        delimiter = None if recursive else "/"

        try:
            blobs = self._client.list_blobs(
                self._bucket.name,
                prefix=key_prefix,
                delimiter=delimiter,
            )
            items = [self._file_info_from_blob(blob) for blob in blobs]
        except Exception as exc:
            raise StorageListError(
                f"Failed to list objects at prefix {prefix} in bucket {self._bucket.name}: {exc}"
            ) from exc

        items.sort(key=lambda f: f.path)
        return items

    async def exists(
        self,
        remote_path: str,
    ) -> bool:
        key = self._key(remote_path)
        blob = self._bucket.blob(key)
        try:
            result: bool = blob.exists()
            return result
        except Exception as exc:
            raise StorageConnectionError(
                f"Failed to check existence of gs://{self._bucket.name}/{key}: {exc}"
            ) from exc

    async def presigned_url(
        self,
        remote_path: str,
        *,
        expiration: int = 3600,
    ) -> str:
        key = self._key(remote_path)
        blob = self._bucket.blob(key)
        if not blob.exists():
            raise StorageNotFoundError(f"Object does not exist: gs://{self._bucket.name}/{key}")
        try:
            url: str = blob.generate_signed_url(
                expiration=expiration,
                method="GET",
            )
            return url
        except Exception as exc:
            raise StorageConnectionError(
                f"Failed to generate signed URL for gs://{self._bucket.name}/{key}: {exc}"
            ) from exc

LocalFilesystemAdapter

Storage adapter backed by the local filesystem.

Designed for development and testing. All remote paths are resolved relative to the configured base_path.

Source code in src/forge/storage/adapters/local.py

class LocalFilesystemAdapter:
    """
    Storage adapter backed by the local filesystem.

    Designed for development and testing. All remote paths are resolved
    relative to the configured *base_path*.
    """

    def __init__(self, base_path: str | Path = "/tmp/forge-storage") -> None:
        self._base_path = Path(base_path).resolve()
        self._base_path.mkdir(parents=True, exist_ok=True)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _resolve(self, remote_path: str) -> Path:
        normalized = normalize_path(remote_path)
        resolved = (self._base_path / normalized).resolve()
        if not str(resolved).startswith(str(self._base_path)):
            raise StoragePermissionError(f"Path traversal detected: {remote_path}")
        return resolved

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def upload(
        self,
        local_path: str,
        remote_path: str,
        *,
        content_type: str | None = None,
        metadata: dict[str, str] | None = None,
        multipart_threshold: int = 100 * 1024 * 1024,
    ) -> FileInfo:
        src = Path(local_path)
        if not src.is_file():
            raise StorageUploadError(f"Local source does not exist or is not a file: {local_path}")

        dst = self._resolve(remote_path)
        dst.parent.mkdir(parents=True, exist_ok=True)

        try:
            shutil.copy2(str(src), str(dst))
        except OSError as exc:
            raise StorageUploadError(
                f"Failed to upload {local_path} -> {remote_path}: {exc}"
            ) from exc

        ct = content_type or detect_content_type(local_path)
        stat = dst.stat()
        return FileInfo(
            path=normalize_path(remote_path),
            size=stat.st_size,
            last_modified=datetime.fromtimestamp(stat.st_mtime, tz=UTC),
            content_type=ct,
            metadata=metadata or {},
        )

    async def download(
        self,
        remote_path: str,
        local_path: str,
        *,
        overwrite: bool = False,
    ) -> str:
        src = self._resolve(remote_path)
        if not src.is_file():
            raise StorageNotFoundError(f"Remote path does not exist: {remote_path}")

        dst = Path(local_path)
        if dst.exists() and not overwrite:
            raise StorageDownloadError(
                f"Local file already exists: {local_path} (use overwrite=True to replace)"
            )
        dst.parent.mkdir(parents=True, exist_ok=True)

        try:
            shutil.copy2(str(src), str(dst))
        except OSError as exc:
            raise StorageDownloadError(
                f"Failed to download {remote_path} -> {local_path}: {exc}"
            ) from exc

        return str(dst.resolve())

    async def delete(
        self,
        remote_path: str,
    ) -> bool:
        path = self._resolve(remote_path)
        if not path.exists():
            return False
        try:
            if path.is_file():
                path.unlink()
            elif path.is_dir():
                shutil.rmtree(str(path))
            else:
                return False
        except OSError as exc:
            raise StorageDeleteError(f"Failed to delete {remote_path}: {exc}") from exc
        return True

    async def list(
        self,
        prefix: str = "",
        *,
        recursive: bool = True,
    ) -> list[FileInfo]:
        resolved = self._resolve(prefix)
        if not resolved.exists():
            return []

        items: list[FileInfo] = []
        pattern = "**/*" if recursive else "*"

        try:
            for entry in resolved.glob(pattern):
                if not entry.is_file():
                    continue
                rel_path = entry.relative_to(self._base_path).as_posix()
                stat = entry.stat()
                ct, _ = mimetypes.guess_type(str(entry))
                items.append(
                    FileInfo(
                        path=rel_path,
                        size=stat.st_size,
                        last_modified=datetime.fromtimestamp(stat.st_mtime, tz=UTC),
                        content_type=ct or "application/octet-stream",
                    )
                )
        except OSError as exc:
            raise StorageListError(f"Failed to list objects at prefix {prefix}: {exc}") from exc

        items.sort(key=lambda f: f.path)
        return items

    async def exists(
        self,
        remote_path: str,
    ) -> bool:
        return self._resolve(remote_path).exists()

    async def presigned_url(
        self,
        remote_path: str,
        *,
        expiration: int = 3600,
    ) -> str:
        resolved = self._resolve(remote_path)
        if not resolved.is_file():
            raise StorageNotFoundError(f"Remote path does not exist: {remote_path}")
        return resolved.as_uri()

S3Adapter

Storage adapter backed by AWS S3 (or S3-compatible) object storage.

Uses boto3 under the hood. Multipart upload is automatically used for files above multipart_threshold.

Source code in src/forge/storage/adapters/s3.py

class S3Adapter:
    """
    Storage adapter backed by AWS S3 (or S3-compatible) object storage.

    Uses *boto3* under the hood. Multipart upload is automatically used
    for files above *multipart_threshold*.
    """

    def __init__(
        self,
        bucket: str,
        *,
        region: str | None = None,
        endpoint_url: str | None = None,
        access_key_id: str | None = None,
        secret_access_key: str | None = None,
        session_token: str | None = None,
        **kwargs: Any,
    ) -> None:
        import boto3

        self._bucket = bucket
        session_kwargs: dict[str, Any] = {}
        if region:
            session_kwargs["region_name"] = region
        if access_key_id and secret_access_key:
            session_kwargs["aws_access_key_id"] = access_key_id
            session_kwargs["aws_secret_access_key"] = secret_access_key
        if session_token:
            session_kwargs["aws_session_token"] = session_token

        session = boto3.Session(**session_kwargs)

        client_kwargs: dict[str, Any] = {}
        if endpoint_url:
            client_kwargs["endpoint_url"] = endpoint_url
        client_kwargs.update(kwargs)

        self._client = session.client("s3", **client_kwargs)

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _key(self, remote_path: str) -> str:
        return normalize_path(remote_path)

    def _file_info_from_head(self, remote_path: str, head: dict[str, Any]) -> FileInfo:
        ct = head.get("ContentType") or "application/octet-stream"
        last_mod = head.get("LastModified")
        return FileInfo(
            path=normalize_path(remote_path),
            size=head.get("ContentLength", 0),
            last_modified=last_mod.replace(tzinfo=UTC) if last_mod else None,
            content_type=ct,
            etag=head.get("ETag", "").strip('"'),
            metadata=head.get("Metadata", {}),
        )

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def upload(
        self,
        local_path: str,
        remote_path: str,
        *,
        content_type: str | None = None,
        metadata: dict[str, str] | None = None,
        multipart_threshold: int = 100 * 1024 * 1024,
    ) -> FileInfo:
        key = self._key(remote_path)
        ct = content_type or detect_content_type(local_path)

        extra_args: dict[str, Any] = {}
        if ct:
            extra_args["ContentType"] = ct
        if metadata:
            extra_args["Metadata"] = metadata

        file_size = Path(local_path).stat().st_size
        config_kwargs: dict[str, Any] = {}
        if file_size > multipart_threshold:
            config_kwargs = {
                "multipart_threshold": multipart_threshold,
                "multipart_chunksize": min(multipart_threshold // 2, 50 * 1024 * 1024),
            }

        try:
            transfer_config = TransferConfig(**config_kwargs) if config_kwargs else None
            self._client.upload_file(
                local_path,
                self._bucket,
                key,
                ExtraArgs=extra_args or None,
                Config=transfer_config,
            )
        except Exception as exc:
            raise StorageUploadError(
                f"Failed to upload {local_path} to s3://{self._bucket}/{key}: {exc}"
            ) from exc

        head = self._client.head_object(Bucket=self._bucket, Key=key)
        return self._file_info_from_head(remote_path, head)

    async def download(
        self,
        remote_path: str,
        local_path: str,
        *,
        overwrite: bool = False,
    ) -> str:
        key = self._key(remote_path)
        dst = Path(local_path)

        if dst.exists() and not overwrite:
            raise StorageDownloadError(
                f"Local file already exists: {local_path} (use overwrite=True to replace)"
            )

        dst.parent.mkdir(parents=True, exist_ok=True)

        try:
            self._client.download_file(self._bucket, key, local_path)
        except self._client.exceptions.NoSuchKey as exc:
            raise StorageNotFoundError(f"Object does not exist: s3://{self._bucket}/{key}") from exc
        except Exception as exc:
            raise StorageDownloadError(
                f"Failed to download s3://{self._bucket}/{key} to {local_path}: {exc}"
            ) from exc

        return str(dst.resolve())

    async def delete(
        self,
        remote_path: str,
    ) -> bool:
        key = self._key(remote_path)
        if not await self.exists(remote_path):
            return False
        try:
            self._client.delete_object(Bucket=self._bucket, Key=key)
            return True
        except Exception as exc:
            raise StorageDeleteError(f"Failed to delete s3://{self._bucket}/{key}: {exc}") from exc

    async def list(
        self,
        prefix: str = "",
        *,
        recursive: bool = True,
    ) -> list[FileInfo]:
        key_prefix = self._key(prefix)
        items: list[FileInfo] = []

        paginator = self._client.get_paginator("list_objects_v2")
        try:
            page_iterator = paginator.paginate(Bucket=self._bucket, Prefix=key_prefix)
            for page in page_iterator:
                contents = page.get("Contents", [])
                for obj in contents:
                    obj_key: str = obj["Key"]
                    if not recursive and "/" in obj_key[len(key_prefix) :].lstrip("/"):
                        continue
                    last_mod = obj.get("LastModified")
                    items.append(
                        FileInfo(
                            path=obj_key,
                            size=obj.get("Size", 0),
                            last_modified=last_mod.replace(tzinfo=UTC) if last_mod else None,
                            etag=obj.get("ETag", "").strip('"'),
                        )
                    )
        except Exception as exc:
            raise StorageListError(
                f"Failed to list objects at prefix {prefix} in bucket {self._bucket}: {exc}"
            ) from exc

        items.sort(key=lambda f: f.path)
        return items

    async def exists(
        self,
        remote_path: str,
    ) -> bool:
        key = self._key(remote_path)
        try:
            self._client.head_object(Bucket=self._bucket, Key=key)
            return True
        except self._client.exceptions.ClientError as exc:
            code = exc.response["Error"]["Code"]
            if code in {"404", "NoSuchKey"}:
                return False
            raise StorageConnectionError(
                f"Failed to check existence of s3://{self._bucket}/{key}: {exc}"
            ) from exc

    async def presigned_url(
        self,
        remote_path: str,
        *,
        expiration: int = 3600,
    ) -> str:
        key = self._key(remote_path)
        if not await self.exists(remote_path):
            raise StorageNotFoundError(f"Object does not exist: s3://{self._bucket}/{key}")
        try:
            url: str = self._client.generate_presigned_url(
                "get_object",
                Params={"Bucket": self._bucket, "Key": key},
                ExpiresIn=expiration,
            )
            return url
        except Exception as exc:
            raise StorageConnectionError(
                f"Failed to generate presigned URL for s3://{self._bucket}/{key}: {exc}"
            ) from exc

StorageBackend

Bases: Protocol

Protocol that every storage adapter must implement.

Source code in src/forge/storage/adapters/base.py

@runtime_checkable
class StorageBackend(Protocol):
    """Protocol that every storage adapter must implement."""

    @abc.abstractmethod
    async def upload(
        self,
        local_path: str,
        remote_path: str,
        *,
        content_type: str | None = None,
        metadata: dict[str, str] | None = None,
        multipart_threshold: int = 100 * 1024 * 1024,
    ) -> FileInfo: ...

    @abc.abstractmethod
    async def download(
        self,
        remote_path: str,
        local_path: str,
        *,
        overwrite: bool = False,
    ) -> str: ...

    @abc.abstractmethod
    async def delete(
        self,
        remote_path: str,
    ) -> bool: ...

    @abc.abstractmethod
    async def list(
        self,
        prefix: str = "",
        *,
        recursive: bool = True,
    ) -> list[FileInfo]: ...

    @abc.abstractmethod
    async def exists(
        self,
        remote_path: str,
    ) -> bool: ...

    @abc.abstractmethod
    async def presigned_url(
        self,
        remote_path: str,
        *,
        expiration: int = 3600,
    ) -> str: ...

StorageConnectionError

Bases: StorageError

Raised when a connection to a storage backend fails.

Source code in src/forge/storage/exceptions.py

class StorageConnectionError(StorageError):
    """Raised when a connection to a storage backend fails."""

StorageDeleteError

Bases: StorageError

Raised when a delete operation fails.

Source code in src/forge/storage/exceptions.py

class StorageDeleteError(StorageError):
    """Raised when a delete operation fails."""

StorageDownloadError

Bases: StorageError

Raised when a download operation fails.

Source code in src/forge/storage/exceptions.py

class StorageDownloadError(StorageError):
    """Raised when a download operation fails."""

StorageError

Bases: ForgeError

Base exception for all storage-related errors.

Source code in src/forge/storage/exceptions.py

class StorageError(ForgeError):
    """Base exception for all storage-related errors."""

StorageListError

Bases: StorageError

Raised when a list operation fails.

Source code in src/forge/storage/exceptions.py

class StorageListError(StorageError):
    """Raised when a list operation fails."""

StorageModule

Bases: ForgeModule

Manages object storage services with pluggable backend adapters.

Source code in src/forge/storage/module.py

class StorageModule(ForgeModule):
    """Manages object storage services with pluggable backend adapters."""

    name = "storage"
    dependencies: ClassVar[list[str]] = ["config"]

    def __init__(self) -> None:
        super().__init__()
        self._backend: StorageBackend | None = None
        self._runtime: Runtime | None = None

    @property
    def backend(self) -> StorageBackend:
        if self._backend is None:
            raise RuntimeError("Storage module is not initialized.")
        return self._backend

    async def setup(self, runtime: Runtime) -> None:
        self._runtime = runtime

        from forge.config.module import ConfigModule

        config_module = cast("ConfigModule", runtime.get(ConfigModule))
        config = getattr(config_module.config, "storage", None)

        backend_type = "local"
        local_base_path = "/tmp/forge-storage"
        s3_bucket: str | None = None
        s3_region: str | None = None
        s3_endpoint: str | None = None
        gcs_bucket: str | None = None
        gcs_project: str | None = None
        gcs_credentials: str | None = None
        azure_container: str | None = None
        azure_connection_string: str | None = None

        if config is not None:
            backend_type = getattr(config, "backend", "local")
            local_base_path = getattr(config, "local_base_path", "/tmp/forge-storage")
            s3_config = getattr(config, "s3", None)
            if s3_config is not None:
                s3_bucket = getattr(s3_config, "bucket", None)
                s3_region = getattr(s3_config, "region", None)
                s3_endpoint = getattr(s3_config, "endpoint_url", None)
            gcs_config = getattr(config, "gcs", None)
            if gcs_config is not None:
                gcs_bucket = getattr(gcs_config, "bucket", None)
                gcs_project = getattr(gcs_config, "project", None)
                gcs_credentials = getattr(gcs_config, "credentials_path", None)
            azure_config = getattr(config, "azure", None)
            if azure_config is not None:
                azure_container = getattr(azure_config, "container", None)
                azure_connection_string = getattr(azure_config, "connection_string", None)

        if backend_type == "s3":
            from forge.storage.adapters.s3 import S3Adapter

            if not s3_bucket:
                raise StorageError("S3 backend requires a 'bucket' configuration.")
            self._backend = S3Adapter(
                bucket=s3_bucket,
                region=s3_region,
                endpoint_url=s3_endpoint,
            )
        elif backend_type == "gcs":
            from forge.storage.adapters.gcs import GCSAdapter

            if not gcs_bucket:
                raise StorageError("GCS backend requires a 'bucket' configuration.")
            self._backend = GCSAdapter(
                bucket=gcs_bucket,
                project=gcs_project,
                credentials_path=gcs_credentials,
            )
        elif backend_type == "azure":
            from forge.storage.adapters.azure import AzureAdapter

            if not azure_container:
                raise StorageError("Azure backend requires a 'container' configuration.")
            self._backend = AzureAdapter(
                container=azure_container,
                connection_string=azure_connection_string,
            )
        else:
            self._backend = LocalFilesystemAdapter(base_path=local_base_path)

    async def teardown(self) -> None:
        self._backend = None
        self._runtime = None

    # ------------------------------------------------------------------
    # Public API — delegates to the active backend
    # ------------------------------------------------------------------

    async def upload(
        self,
        local_path: str,
        remote_path: str,
        *,
        content_type: str | None = None,
        metadata: dict[str, str] | None = None,
        multipart_threshold: int = 100 * 1024 * 1024,
    ) -> FileInfo:
        return await self.backend.upload(
            local_path,
            remote_path,
            content_type=content_type,
            metadata=metadata,
            multipart_threshold=multipart_threshold,
        )

    async def download(
        self,
        remote_path: str,
        local_path: str,
        *,
        overwrite: bool = False,
    ) -> str:
        return await self.backend.download(remote_path, local_path, overwrite=overwrite)

    async def delete(self, remote_path: str) -> bool:
        return await self.backend.delete(remote_path)

    async def list(
        self,
        prefix: str = "",
        *,
        recursive: bool = True,
    ) -> list[FileInfo]:
        return await self.backend.list(prefix, recursive=recursive)

    async def exists(self, remote_path: str) -> bool:
        return await self.backend.exists(remote_path)

    async def presigned_url(
        self,
        remote_path: str,
        *,
        expiration: int = 3600,
    ) -> str:
        return await self.backend.presigned_url(remote_path, expiration=expiration)

    def health_check(self) -> HealthResult:
        backend = self._backend
        if backend is None:
            return HealthResult.error("Storage backend not initialized")

        if isinstance(backend, LocalFilesystemAdapter):
            return HealthResult(HealthResult.OK, "Local filesystem storage active")

        try:
            from forge.core.async_bridge import run_async_health_check

            async def _ping() -> None:
                await backend.exists("health-check-probe")

            run_async_health_check(_ping())
            return HealthResult.ok()
        except Exception as exc:
            return HealthResult.error(f"Storage backend health check failed: {exc}")

StorageNotFoundError

Bases: StorageError

Raised when a remote path does not exist.

Source code in src/forge/storage/exceptions.py

class StorageNotFoundError(StorageError):
    """Raised when a remote path does not exist."""

StoragePermissionError

Bases: StorageError

Raised when access to a storage resource is denied.

Source code in src/forge/storage/exceptions.py

class StoragePermissionError(StorageError):
    """Raised when access to a storage resource is denied."""

StorageUploadError

Bases: StorageError

Raised when an upload operation fails.

Source code in src/forge/storage/exceptions.py

class StorageUploadError(StorageError):
    """Raised when an upload operation fails."""

Functions:

normalize_path

normalize_path(path: str) -> str

Normalize a storage path to use forward slashes and strip leading slash.

Source code in src/forge/storage/adapters/base.py

def normalize_path(path: str) -> str:
    """Normalize a storage path to use forward slashes and strip leading slash."""
    normalized = path.replace("\\", "/").replace(os.sep, "/").strip("/")
    return normalized