Abstract interface

The abstract base classes that define the unified storage contract. Both the AWS S3 and GCS implementations derive from these, so you can write code against the abstract types to stay provider-agnostic.

unicloud.abstract_class.CloudStorageFactory

Bases: ABC

Provider-level cloud-storage client contract.

Every concrete provider (AWS S3, Google Cloud Storage, ...) must subclass this ABC and implement :meth:create_client, :meth:client, :meth:upload, :meth:download, and :meth:get_bucket. Callers that want to stay provider-agnostic should accept CloudStorageFactory and only call the methods declared here.

Examples:

• Accept either provider in a type-annotated function:

  >>> from unicloud.abstract_class import CloudStorageFactory
  >>> def sync(client: CloudStorageFactory, bucket: str, src: str, dst: str) -> None:
  ...     client.get_bucket(bucket).upload(src, dst)
  

Source code in src/unicloud/abstract_class.py

class CloudStorageFactory(ABC):
    """Provider-level cloud-storage client contract.

    Every concrete provider (AWS S3, Google Cloud Storage, ...) must subclass this ABC and implement
    :meth:`create_client`, :meth:`client`, :meth:`upload`, :meth:`download`, and :meth:`get_bucket`.
    Callers that want to stay provider-agnostic should accept ``CloudStorageFactory`` and only call
    the methods declared here.

    Examples:
        - Accept either provider in a type-annotated function:
            ```python
            >>> from unicloud.abstract_class import CloudStorageFactory
            >>> def sync(client: CloudStorageFactory, bucket: str, src: str, dst: str) -> None:
            ...     client.get_bucket(bucket).upload(src, dst)

            ```
    """

    @abstractmethod
    def create_client(self):
        """Construct and return the underlying provider SDK client.

        Implementations read credentials from environment variables (or accept a provider-specific
        config argument) and return the native client object — for example a ``boto3.client`` for AWS
        or a ``google.cloud.storage.Client`` for GCS.

        Returns:
            The provider-specific SDK client instance. Type varies by provider.
        """
        pass

    @property
    @abstractmethod
    def client(self):
        """Return the cached provider SDK client.

        Returns:
            The same instance that :meth:`create_client` produced. Type varies by provider.
        """
        pass

    @abstractmethod
    def upload(self, file_path, destination):
        """Upload a single local file to the provider.

        This is a convenience helper on the factory; prefer :meth:`get_bucket` + the bucket-level
        ``upload`` for anything non-trivial (overwrite handling, directory uploads, etc.).

        Args:
            file_path: Path to the local file to upload.
            destination: Destination in the provider. The convention used by both shipped providers
                is ``"<bucket_name>/<object_key>"``.
        """
        pass

    @abstractmethod
    def download(self, source, file_path):
        """Download a single object to a local path.

        Args:
            source: Provider-side path. Both shipped providers use ``"<bucket_name>/<object_key>"``.
            file_path: Local destination path for the downloaded file.
        """
        pass

    @abstractmethod
    def get_bucket(self, bucket_name) -> "AbstractBucket":
        """Return a bucket handle for per-object operations.

        Args:
            bucket_name: The name (or ID) of the bucket to look up.

        Returns:
            AbstractBucket: A concrete :class:`AbstractBucket` subclass from the provider.
        """
        pass

client abstractmethod property

Return the cached provider SDK client.

Returns:

Type	Description
	The same instance that :meth:create_client produced. Type varies by provider.

create_client() abstractmethod

Construct and return the underlying provider SDK client.

Implementations read credentials from environment variables (or accept a provider-specific config argument) and return the native client object — for example a boto3.client for AWS or a google.cloud.storage.Client for GCS.

Returns:

Type	Description
	The provider-specific SDK client instance. Type varies by provider.

Source code in src/unicloud/abstract_class.py

@abstractmethod
def create_client(self):
    """Construct and return the underlying provider SDK client.

    Implementations read credentials from environment variables (or accept a provider-specific
    config argument) and return the native client object — for example a ``boto3.client`` for AWS
    or a ``google.cloud.storage.Client`` for GCS.

    Returns:
        The provider-specific SDK client instance. Type varies by provider.
    """
    pass

upload(file_path, destination) abstractmethod

Upload a single local file to the provider.

This is a convenience helper on the factory; prefer :meth:get_bucket + the bucket-level upload for anything non-trivial (overwrite handling, directory uploads, etc.).

Parameters:

Name	Type	Description	Default
file_path		Path to the local file to upload.	required
destination		Destination in the provider. The convention used by both shipped providers is "<bucket_name>/<object_key>".	required

Source code in src/unicloud/abstract_class.py

@abstractmethod
def upload(self, file_path, destination):
    """Upload a single local file to the provider.

    This is a convenience helper on the factory; prefer :meth:`get_bucket` + the bucket-level
    ``upload`` for anything non-trivial (overwrite handling, directory uploads, etc.).

    Args:
        file_path: Path to the local file to upload.
        destination: Destination in the provider. The convention used by both shipped providers
            is ``"<bucket_name>/<object_key>"``.
    """
    pass

download(source, file_path) abstractmethod

Download a single object to a local path.

Parameters:

Name	Type	Description	Default
source		Provider-side path. Both shipped providers use "<bucket_name>/<object_key>".	required
file_path		Local destination path for the downloaded file.	required

Source code in src/unicloud/abstract_class.py

@abstractmethod
def download(self, source, file_path):
    """Download a single object to a local path.

    Args:
        source: Provider-side path. Both shipped providers use ``"<bucket_name>/<object_key>"``.
        file_path: Local destination path for the downloaded file.
    """
    pass

get_bucket(bucket_name) abstractmethod

Return a bucket handle for per-object operations.

Parameters:

Name	Type	Description	Default
bucket_name		The name (or ID) of the bucket to look up.	required

Returns:

Name	Type	Description
AbstractBucket	AbstractBucket	A concrete :class:AbstractBucket subclass from the provider.

Source code in src/unicloud/abstract_class.py

@abstractmethod
def get_bucket(self, bucket_name) -> "AbstractBucket":
    """Return a bucket handle for per-object operations.

    Args:
        bucket_name: The name (or ID) of the bucket to look up.

    Returns:
        AbstractBucket: A concrete :class:`AbstractBucket` subclass from the provider.
    """
    pass

unicloud.abstract_class.AbstractBucket

Bases: ABC

Per-bucket operation contract.

A bucket handle exposes upload/download/delete/list/exists for a single bucket. Both providers ship a :class:Bucket class that subclasses this ABC and shares the same method names; provider parity is intentional.

Examples:

• Type-annotate a helper that works with either provider's bucket:

  >>> from unicloud.abstract_class import AbstractBucket
  >>> def push_then_verify(bucket: AbstractBucket, local: str, remote: str) -> bool:
  ...     bucket.upload(local, remote)
  ...     return bucket.file_exists(remote)
  

Source code in src/unicloud/abstract_class.py

class AbstractBucket(ABC):
    """Per-bucket operation contract.

    A bucket handle exposes upload/download/delete/list/exists for a single bucket. Both providers
    ship a :class:`Bucket` class that subclasses this ABC and shares the same method names; provider
    parity is intentional.

    Examples:
        - Type-annotate a helper that works with either provider's bucket:
            ```python
            >>> from unicloud.abstract_class import AbstractBucket
            >>> def push_then_verify(bucket: AbstractBucket, local: str, remote: str) -> bool:
            ...     bucket.upload(local, remote)
            ...     return bucket.file_exists(remote)

            ```
    """

    @abstractmethod
    def __str__(self):
        """Return a short, human-readable representation of the bucket.

        Returns:
            str: Usually ``"Bucket: <name>"``.
        """
        pass

    @abstractmethod
    def __repr__(self):
        """Return a developer-facing representation of the bucket.

        Returns:
            str: Usually the same as :meth:`__str__`.
        """
        pass

    @abstractmethod
    def upload(
        self,
        local_path: Union[str, Path],
        bucket_path: Union[str, Path],
        overwrite: bool = False,
    ):
        """Upload a file or directory into the bucket.

        Implementations must accept both single files and directories; when ``local_path`` is a
        directory, the upload is recursive and preserves the relative tree.

        Args:
            local_path: Local file or directory to upload.
            bucket_path: Destination prefix inside the bucket.
            overwrite: If ``False`` (the default), existing destination objects cause a ``ValueError``.
                If ``True``, existing objects are overwritten silently.
        """
        pass

    @abstractmethod
    def download(
        self, bucket_path: str, local_path: Union[str, Path], overwrite: bool = False
    ):
        """Download a file or directory out of the bucket.

        A trailing ``/`` on ``bucket_path`` signals a directory download; otherwise a single object
        is fetched.

        Args:
            bucket_path: Source path inside the bucket. Trailing ``/`` triggers a recursive download.
            local_path: Local destination path.
            overwrite: If ``False`` (the default), existing local files cause a ``ValueError``.
                If ``True``, existing files are overwritten.
        """
        pass

    @abstractmethod
    def delete(self, bucket_path: str):
        """Delete a file or directory from the bucket.

        A trailing ``/`` on ``bucket_path`` signals a recursive directory delete.

        Args:
            bucket_path: Path inside the bucket to delete.
        """
        pass

    @abstractmethod
    def list_files(self):
        """List the objects in the bucket.

        Implementations may accept provider-specific filtering arguments (prefix, glob pattern,
        max results, ...) as additional parameters.

        Returns:
            list[str]: Object keys, in provider-defined order.
        """
        pass

    @abstractmethod
    def file_exists(self, file_name: str) -> bool:
        """Return whether an object exists in the bucket.

        Args:
            file_name: Object key to check.

        Returns:
            bool: ``True`` if the object exists, ``False`` otherwise.
        """
        pass

    @property
    @abstractmethod
    def name(self):
        """Return the bucket name.

        Returns:
            str: The bucket's name (AWS) or ID (GCS).
        """
        pass

name abstractmethod property

Return the bucket name.

Returns:

Name	Type	Description
str		The bucket's name (AWS) or ID (GCS).

__str__() abstractmethod

Return a short, human-readable representation of the bucket.

Returns:

Name	Type	Description
str		Usually "Bucket: <name>".

Source code in src/unicloud/abstract_class.py

@abstractmethod
def __str__(self):
    """Return a short, human-readable representation of the bucket.

    Returns:
        str: Usually ``"Bucket: <name>"``.
    """
    pass

__repr__() abstractmethod

Return a developer-facing representation of the bucket.

Returns:

Name	Type	Description
str		Usually the same as :meth:__str__.

Source code in src/unicloud/abstract_class.py

@abstractmethod
def __repr__(self):
    """Return a developer-facing representation of the bucket.

    Returns:
        str: Usually the same as :meth:`__str__`.
    """
    pass

upload(local_path, bucket_path, overwrite=False) abstractmethod

Upload a file or directory into the bucket.

Implementations must accept both single files and directories; when local_path is a directory, the upload is recursive and preserves the relative tree.

Parameters:

Name	Type	Description	Default
local_path	Union[str, Path]	Local file or directory to upload.	required
bucket_path	Union[str, Path]	Destination prefix inside the bucket.	required
overwrite	bool	If False (the default), existing destination objects cause a ValueError. If True, existing objects are overwritten silently.	False

Source code in src/unicloud/abstract_class.py

@abstractmethod
def upload(
    self,
    local_path: Union[str, Path],
    bucket_path: Union[str, Path],
    overwrite: bool = False,
):
    """Upload a file or directory into the bucket.

    Implementations must accept both single files and directories; when ``local_path`` is a
    directory, the upload is recursive and preserves the relative tree.

    Args:
        local_path: Local file or directory to upload.
        bucket_path: Destination prefix inside the bucket.
        overwrite: If ``False`` (the default), existing destination objects cause a ``ValueError``.
            If ``True``, existing objects are overwritten silently.
    """
    pass

download(bucket_path, local_path, overwrite=False) abstractmethod

Download a file or directory out of the bucket.

A trailing / on bucket_path signals a directory download; otherwise a single object is fetched.

Parameters:

Name	Type	Description	Default
bucket_path	str	Source path inside the bucket. Trailing / triggers a recursive download.	required
local_path	Union[str, Path]	Local destination path.	required
overwrite	bool	If False (the default), existing local files cause a ValueError. If True, existing files are overwritten.	False

Source code in src/unicloud/abstract_class.py

@abstractmethod
def download(
    self, bucket_path: str, local_path: Union[str, Path], overwrite: bool = False
):
    """Download a file or directory out of the bucket.

    A trailing ``/`` on ``bucket_path`` signals a directory download; otherwise a single object
    is fetched.

    Args:
        bucket_path: Source path inside the bucket. Trailing ``/`` triggers a recursive download.
        local_path: Local destination path.
        overwrite: If ``False`` (the default), existing local files cause a ``ValueError``.
            If ``True``, existing files are overwritten.
    """
    pass

delete(bucket_path) abstractmethod

Delete a file or directory from the bucket.

A trailing / on bucket_path signals a recursive directory delete.

Parameters:

Name	Type	Description	Default
bucket_path	str	Path inside the bucket to delete.	required

Source code in src/unicloud/abstract_class.py

@abstractmethod
def delete(self, bucket_path: str):
    """Delete a file or directory from the bucket.

    A trailing ``/`` on ``bucket_path`` signals a recursive directory delete.

    Args:
        bucket_path: Path inside the bucket to delete.
    """
    pass

list_files() abstractmethod

List the objects in the bucket.

Implementations may accept provider-specific filtering arguments (prefix, glob pattern, max results, ...) as additional parameters.

Returns:

Type	Description
	list[str]: Object keys, in provider-defined order.

Source code in src/unicloud/abstract_class.py

@abstractmethod
def list_files(self):
    """List the objects in the bucket.

    Implementations may accept provider-specific filtering arguments (prefix, glob pattern,
    max results, ...) as additional parameters.

    Returns:
        list[str]: Object keys, in provider-defined order.
    """
    pass

file_exists(file_name) abstractmethod

Return whether an object exists in the bucket.

Parameters:

Name	Type	Description	Default
file_name	str	Object key to check.	required

Returns:

Name	Type	Description
bool	bool	True if the object exists, False otherwise.

Source code in src/unicloud/abstract_class.py

@abstractmethod
def file_exists(self, file_name: str) -> bool:
    """Return whether an object exists in the bucket.

    Args:
        file_name: Object key to check.

    Returns:
        bool: ``True`` if the object exists, ``False`` otherwise.
    """
    pass