MinIO

The minio adapter provides integration with MinIO (and S3-compatible object storage) for uploading, downloading, and managing binary objects.

Ports

Abstract port interface defining the MinIO adapter contract for object storage operations.

MinIO port definitions for ArchiPy.

archipy.adapters.minio.ports.MinioObjectType module-attribute

MinioObjectType = dict[str, Any]

archipy.adapters.minio.ports.MinioBucketType module-attribute

MinioBucketType = dict[str, Any]

archipy.adapters.minio.ports.MinioPolicyType module-attribute

MinioPolicyType = dict[str, Any]

archipy.adapters.minio.ports.MinioLifecycleRuleType module-attribute

MinioLifecycleRuleType = dict[str, Any]

archipy.adapters.minio.ports.MinioObjectVersionType module-attribute

MinioObjectVersionType = dict[str, Any]

archipy.adapters.minio.ports.MinioPort

Interface for MinIO operations providing a standardized access pattern.

This interface defines the contract for MinIO adapters, ensuring consistent implementation of object storage operations across different adapters.

Source code in archipy/adapters/minio/ports.py

class MinioPort:
    """Interface for MinIO operations providing a standardized access pattern.

    This interface defines the contract for MinIO adapters, ensuring consistent
    implementation of object storage operations across different adapters.
    """

    # Bucket Operations
    @abstractmethod
    def bucket_exists(self, bucket_name: str) -> bool:
        """Check if a bucket exists."""
        raise NotImplementedError

    @abstractmethod
    def make_bucket(self, bucket_name: str) -> None:
        """Create a new bucket."""
        raise NotImplementedError

    @abstractmethod
    def remove_bucket(self, bucket_name: str) -> None:
        """Remove a bucket."""
        raise NotImplementedError

    @abstractmethod
    def list_buckets(self) -> list[MinioBucketType]:
        """List all buckets."""
        raise NotImplementedError

    # Object Operations
    @abstractmethod
    def put_object(
        self,
        bucket_name: str,
        object_name: str,
        file_path: str,
        tags: dict[str, str] | None = None,
    ) -> None:
        """Upload a file to a bucket.

        Args:
            bucket_name: Destination bucket name.
            object_name: Object name in the bucket.
            file_path: Local file path to upload.
            tags: Optional key-value tags to attach to the object. Tags can be used
                with bucket lifecycle rules to implement per-object TTL expiration.
        """
        raise NotImplementedError

    @abstractmethod
    def get_object(self, bucket_name: str, object_name: str, file_path: str) -> None:
        """Download an object to a file."""
        raise NotImplementedError

    @abstractmethod
    def remove_object(self, bucket_name: str, object_name: str) -> None:
        """Remove an object from a bucket."""
        raise NotImplementedError

    @abstractmethod
    def remove_objects(self, bucket_name: str, object_names: list[str]) -> None:
        """Remove multiple objects from a bucket in a single request.

        Args:
            bucket_name: Bucket name.
            object_names: Object keys to delete.

        Raises:
            InvalidArgumentError: If bucket_name is empty or object_names is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to remove objects is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def object_exists(self, bucket_name: str, object_name: str) -> bool:
        """Check if an object exists in a bucket.

        Args:
            bucket_name: Bucket name.
            object_name: Object key to check.

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

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            PermissionDeniedError: If permission to check the object is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def list_objects(
        self,
        bucket_name: str,
        prefix: str = "",
        *,  # Force recursive to be keyword-only to avoid boolean flag issues
        recursive: bool = False,
    ) -> list[MinioObjectType]:
        """List objects in a bucket.

        Args:
            bucket_name: The name of the bucket to list objects from
            prefix: Optional prefix to filter objects by
            recursive: Whether to list objects recursively (include sub-directories)

        Returns:
            A list of MinioObjectType objects
        """
        raise NotImplementedError

    @abstractmethod
    def stat_object(self, bucket_name: str, object_name: str) -> MinioObjectType:
        """Get object metadata."""
        raise NotImplementedError

    # Presigned URL Operations
    @abstractmethod
    def presigned_get_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
        """Generate a presigned URL for downloading an object."""
        raise NotImplementedError

    @abstractmethod
    def presigned_put_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
        """Generate a presigned URL for uploading an object."""
        raise NotImplementedError

    # Policy Operations
    @abstractmethod
    def set_bucket_policy(self, bucket_name: str, policy: str) -> None:
        """Set bucket policy."""
        raise NotImplementedError

    @abstractmethod
    def get_bucket_policy(self, bucket_name: str) -> MinioPolicyType:
        """Get bucket policy."""
        raise NotImplementedError

    @abstractmethod
    def copy_object(
        self,
        src_bucket_name: str,
        src_object_name: str,
        dest_bucket_name: str,
        dest_object_name: str,
    ) -> None:
        """Copy an object within or between buckets."""
        raise NotImplementedError

    @abstractmethod
    def put_object_stream(
        self,
        bucket_name: str,
        object_name: str,
        data: bytes | BinaryIO,
        length: int = -1,
        content_type: str = "application/octet-stream",
        tags: dict[str, str] | None = None,
    ) -> None:
        """Upload data from a bytes buffer or binary stream to a bucket.

        Unlike put_object which requires a local file path, this method accepts
        in-memory bytes or any binary stream, avoiding the need for a temporary file.

        Args:
            bucket_name: Destination bucket name.
            object_name: Object name in the bucket.
            data: Content to upload as raw bytes or a binary stream (BinaryIO).
            length: Content length in bytes. If -1, computed automatically for bytes;
                for streams, providing the exact length avoids buffering overhead.
            content_type: MIME type of the content. Defaults to "application/octet-stream".
            tags: Optional key-value tags to attach to the object. Tags can be used
                with bucket lifecycle rules to implement per-object TTL expiration.
        """
        raise NotImplementedError

    @abstractmethod
    def get_object_stream(self, bucket_name: str, object_name: str) -> bytes:
        """Download an object and return its content as bytes.

        Unlike get_object which requires a local file path, this method returns
        the object content directly in memory, avoiding a temporary file.

        Args:
            bucket_name: Source bucket name.
            object_name: Object name in the bucket.

        Returns:
            bytes: The full content of the object.
        """
        raise NotImplementedError

    # Object Tag Operations
    @abstractmethod
    def set_object_tags(self, bucket_name: str, object_name: str, tags: dict[str, str]) -> None:
        """Set tags on an existing object, replacing any existing tags.

        Tags are key-value pairs that can be used with bucket lifecycle rules to
        implement TTL-based expiration. For example, tag an object with
        ``{"ttl-days": "7"}`` and create a matching lifecycle rule to auto-delete
        it after 7 days.

        Args:
            bucket_name: Bucket containing the object.
            object_name: Object name to tag.
            tags: Key-value mapping of tag names to values.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to set tags is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def remove_object_tags(self, bucket_name: str, object_name: str) -> None:
        """Remove all tags from an object.

        Args:
            bucket_name: Bucket containing the object.
            object_name: Object name whose tags to remove.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to remove tags is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def get_object_tags(self, bucket_name: str, object_name: str) -> dict[str, str]:
        """Return the tags attached to an object.

        Args:
            bucket_name: Bucket containing the object.
            object_name: Object name whose tags to retrieve.

        Returns:
            dict[str, str]: Mapping of tag key to tag value. Empty dict if no tags.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to get tags is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    # Lifecycle / TTL Operations
    @abstractmethod
    def set_bucket_lifecycle(self, bucket_name: str, rules: list[MinioLifecycleRuleType]) -> None:
        """Set the lifecycle configuration for a bucket.

        Lifecycle rules control automatic expiration (TTL) of objects. Each rule is a
        dict that maps to the S3 ``LifecycleRule`` structure. A minimal expiration rule::

            {
                "ID": "expire-after-7-days",
                "Status": "Enabled",
                "Filter": {"Prefix": "uploads/"},
                "Expiration": {"Days": 7},
            }

        To expire only objects with a specific tag::

            {
                "ID": "expire-tagged-7days",
                "Status": "Enabled",
                "Filter": {"Tag": {"Key": "ttl-days", "Value": "7"}},
                "Expiration": {"Days": 7},
            }

        Args:
            bucket_name: Target bucket name.
            rules: List of lifecycle rule dicts (S3 ``LifecycleRule`` format).

        Raises:
            InvalidArgumentError: If bucket_name is empty or rules is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to set lifecycle is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def get_bucket_lifecycle(self, bucket_name: str) -> list[MinioLifecycleRuleType]:
        """Return the lifecycle rules configured for a bucket.

        Args:
            bucket_name: Bucket name.

        Returns:
            list[MinioLifecycleRuleType]: List of lifecycle rule dicts, or an empty
                list if no lifecycle configuration is set.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to get lifecycle is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def delete_bucket_lifecycle(self, bucket_name: str) -> None:
        """Delete the lifecycle configuration from a bucket.

        After this call the bucket has no lifecycle rules and objects will not be
        automatically expired.

        Args:
            bucket_name: Bucket name.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to delete lifecycle is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    # Versioning Operations
    @abstractmethod
    def set_bucket_versioning(self, bucket_name: str, *, enabled: bool) -> None:
        """Enable or suspend versioning on a bucket.

        Args:
            bucket_name: Bucket name.
            enabled: If True, enable versioning; if False, suspend it.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to set versioning is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def get_bucket_versioning(self, bucket_name: str) -> str:
        """Return the versioning status of a bucket.

        Args:
            bucket_name: Bucket name.

        Returns:
            str: Versioning status — ``Enabled``, ``Suspended``, or empty string if never configured.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to get versioning is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def list_object_versions(
        self,
        bucket_name: str,
        prefix: str = "",
    ) -> list[MinioObjectVersionType]:
        """List all versions and delete markers for objects in a bucket.

        Args:
            bucket_name: Bucket name.
            prefix: Optional prefix to filter objects by.

        Returns:
            list[MinioObjectVersionType]: Version entries with keys, version IDs, and metadata.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to list versions is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

    @abstractmethod
    def remove_object_version(self, bucket_name: str, object_name: str, version_id: str) -> None:
        """Permanently delete a specific object version.

        Args:
            bucket_name: Bucket name.
            object_name: Object key.
            version_id: Version ID to delete permanently.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket, object, or version does not exist.
            PermissionDeniedError: If permission to delete the version is denied.
            StorageError: If there's a storage-related error.
        """
        raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.bucket_exists abstractmethod

bucket_exists(bucket_name: str) -> bool

Check if a bucket exists.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def bucket_exists(self, bucket_name: str) -> bool:
    """Check if a bucket exists."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.make_bucket abstractmethod

make_bucket(bucket_name: str) -> None

Create a new bucket.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def make_bucket(self, bucket_name: str) -> None:
    """Create a new bucket."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.remove_bucket abstractmethod

remove_bucket(bucket_name: str) -> None

Remove a bucket.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def remove_bucket(self, bucket_name: str) -> None:
    """Remove a bucket."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.list_buckets abstractmethod

list_buckets() -> list[MinioBucketType]

List all buckets.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def list_buckets(self) -> list[MinioBucketType]:
    """List all buckets."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.put_object abstractmethod

put_object(
    bucket_name: str,
    object_name: str,
    file_path: str,
    tags: dict[str, str] | None = None,
) -> None

Upload a file to a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Destination bucket name.	required
object_name	str	Object name in the bucket.	required
file_path	str	Local file path to upload.	required
tags	dict[str, str] | None	Optional key-value tags to attach to the object. Tags can be used with bucket lifecycle rules to implement per-object TTL expiration.	None

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def put_object(
    self,
    bucket_name: str,
    object_name: str,
    file_path: str,
    tags: dict[str, str] | None = None,
) -> None:
    """Upload a file to a bucket.

    Args:
        bucket_name: Destination bucket name.
        object_name: Object name in the bucket.
        file_path: Local file path to upload.
        tags: Optional key-value tags to attach to the object. Tags can be used
            with bucket lifecycle rules to implement per-object TTL expiration.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.get_object abstractmethod

get_object(
    bucket_name: str, object_name: str, file_path: str
) -> None

Download an object to a file.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def get_object(self, bucket_name: str, object_name: str, file_path: str) -> None:
    """Download an object to a file."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.remove_object abstractmethod

remove_object(bucket_name: str, object_name: str) -> None

Remove an object from a bucket.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def remove_object(self, bucket_name: str, object_name: str) -> None:
    """Remove an object from a bucket."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.remove_objects abstractmethod

remove_objects(
    bucket_name: str, object_names: list[str]
) -> None

Remove multiple objects from a bucket in a single request.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_names	list[str]	Object keys to delete.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty or object_names is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to remove objects is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def remove_objects(self, bucket_name: str, object_names: list[str]) -> None:
    """Remove multiple objects from a bucket in a single request.

    Args:
        bucket_name: Bucket name.
        object_names: Object keys to delete.

    Raises:
        InvalidArgumentError: If bucket_name is empty or object_names is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to remove objects is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.object_exists abstractmethod

object_exists(bucket_name: str, object_name: str) -> bool

Check if an object exists in a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object key to check.	required

Returns:

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

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
PermissionDeniedError	If permission to check the object is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def object_exists(self, bucket_name: str, object_name: str) -> bool:
    """Check if an object exists in a bucket.

    Args:
        bucket_name: Bucket name.
        object_name: Object key to check.

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

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        PermissionDeniedError: If permission to check the object is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.list_objects abstractmethod

list_objects(
    bucket_name: str,
    prefix: str = "",
    *,
    recursive: bool = False,
) -> list[MinioObjectType]

List objects in a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	The name of the bucket to list objects from	required
prefix	str	Optional prefix to filter objects by	''
recursive	bool	Whether to list objects recursively (include sub-directories)	False

Returns:

Type	Description
list[MinioObjectType]	A list of MinioObjectType objects

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def list_objects(
    self,
    bucket_name: str,
    prefix: str = "",
    *,  # Force recursive to be keyword-only to avoid boolean flag issues
    recursive: bool = False,
) -> list[MinioObjectType]:
    """List objects in a bucket.

    Args:
        bucket_name: The name of the bucket to list objects from
        prefix: Optional prefix to filter objects by
        recursive: Whether to list objects recursively (include sub-directories)

    Returns:
        A list of MinioObjectType objects
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.stat_object abstractmethod

stat_object(
    bucket_name: str, object_name: str
) -> MinioObjectType

Get object metadata.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def stat_object(self, bucket_name: str, object_name: str) -> MinioObjectType:
    """Get object metadata."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.presigned_get_object abstractmethod

presigned_get_object(
    bucket_name: str, object_name: str, expires: int = 3600
) -> str

Generate a presigned URL for downloading an object.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def presigned_get_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
    """Generate a presigned URL for downloading an object."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.presigned_put_object abstractmethod

presigned_put_object(
    bucket_name: str, object_name: str, expires: int = 3600
) -> str

Generate a presigned URL for uploading an object.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def presigned_put_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
    """Generate a presigned URL for uploading an object."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.set_bucket_policy abstractmethod

set_bucket_policy(bucket_name: str, policy: str) -> None

Set bucket policy.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def set_bucket_policy(self, bucket_name: str, policy: str) -> None:
    """Set bucket policy."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.get_bucket_policy abstractmethod

get_bucket_policy(bucket_name: str) -> MinioPolicyType

Get bucket policy.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def get_bucket_policy(self, bucket_name: str) -> MinioPolicyType:
    """Get bucket policy."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.copy_object abstractmethod

copy_object(
    src_bucket_name: str,
    src_object_name: str,
    dest_bucket_name: str,
    dest_object_name: str,
) -> None

Copy an object within or between buckets.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def copy_object(
    self,
    src_bucket_name: str,
    src_object_name: str,
    dest_bucket_name: str,
    dest_object_name: str,
) -> None:
    """Copy an object within or between buckets."""
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.put_object_stream abstractmethod

put_object_stream(
    bucket_name: str,
    object_name: str,
    data: bytes | BinaryIO,
    length: int = -1,
    content_type: str = "application/octet-stream",
    tags: dict[str, str] | None = None,
) -> None

Upload data from a bytes buffer or binary stream to a bucket.

Unlike put_object which requires a local file path, this method accepts in-memory bytes or any binary stream, avoiding the need for a temporary file.

Parameters:

Name	Type	Description	Default
bucket_name	str	Destination bucket name.	required
object_name	str	Object name in the bucket.	required
data	bytes | BinaryIO	Content to upload as raw bytes or a binary stream (BinaryIO).	required
length	int	Content length in bytes. If -1, computed automatically for bytes; for streams, providing the exact length avoids buffering overhead.	-1
content_type	str	MIME type of the content. Defaults to "application/octet-stream".	'application/octet-stream'
tags	dict[str, str] | None	Optional key-value tags to attach to the object. Tags can be used with bucket lifecycle rules to implement per-object TTL expiration.	None

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def put_object_stream(
    self,
    bucket_name: str,
    object_name: str,
    data: bytes | BinaryIO,
    length: int = -1,
    content_type: str = "application/octet-stream",
    tags: dict[str, str] | None = None,
) -> None:
    """Upload data from a bytes buffer or binary stream to a bucket.

    Unlike put_object which requires a local file path, this method accepts
    in-memory bytes or any binary stream, avoiding the need for a temporary file.

    Args:
        bucket_name: Destination bucket name.
        object_name: Object name in the bucket.
        data: Content to upload as raw bytes or a binary stream (BinaryIO).
        length: Content length in bytes. If -1, computed automatically for bytes;
            for streams, providing the exact length avoids buffering overhead.
        content_type: MIME type of the content. Defaults to "application/octet-stream".
        tags: Optional key-value tags to attach to the object. Tags can be used
            with bucket lifecycle rules to implement per-object TTL expiration.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.get_object_stream abstractmethod

get_object_stream(
    bucket_name: str, object_name: str
) -> bytes

Download an object and return its content as bytes.

Unlike get_object which requires a local file path, this method returns the object content directly in memory, avoiding a temporary file.

Parameters:

Name	Type	Description	Default
bucket_name	str	Source bucket name.	required
object_name	str	Object name in the bucket.	required

Returns:

Name	Type	Description
bytes	bytes	The full content of the object.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def get_object_stream(self, bucket_name: str, object_name: str) -> bytes:
    """Download an object and return its content as bytes.

    Unlike get_object which requires a local file path, this method returns
    the object content directly in memory, avoiding a temporary file.

    Args:
        bucket_name: Source bucket name.
        object_name: Object name in the bucket.

    Returns:
        bytes: The full content of the object.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.set_object_tags abstractmethod

set_object_tags(
    bucket_name: str, object_name: str, tags: dict[str, str]
) -> None

Set tags on an existing object, replacing any existing tags.

Tags are key-value pairs that can be used with bucket lifecycle rules to implement TTL-based expiration. For example, tag an object with {"ttl-days": "7"} and create a matching lifecycle rule to auto-delete it after 7 days.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket containing the object.	required
object_name	str	Object name to tag.	required
tags	dict[str, str]	Key-value mapping of tag names to values.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to set tags is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def set_object_tags(self, bucket_name: str, object_name: str, tags: dict[str, str]) -> None:
    """Set tags on an existing object, replacing any existing tags.

    Tags are key-value pairs that can be used with bucket lifecycle rules to
    implement TTL-based expiration. For example, tag an object with
    ``{"ttl-days": "7"}`` and create a matching lifecycle rule to auto-delete
    it after 7 days.

    Args:
        bucket_name: Bucket containing the object.
        object_name: Object name to tag.
        tags: Key-value mapping of tag names to values.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to set tags is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.remove_object_tags abstractmethod

remove_object_tags(
    bucket_name: str, object_name: str
) -> None

Remove all tags from an object.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket containing the object.	required
object_name	str	Object name whose tags to remove.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to remove tags is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def remove_object_tags(self, bucket_name: str, object_name: str) -> None:
    """Remove all tags from an object.

    Args:
        bucket_name: Bucket containing the object.
        object_name: Object name whose tags to remove.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to remove tags is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.get_object_tags abstractmethod

get_object_tags(
    bucket_name: str, object_name: str
) -> dict[str, str]

Return the tags attached to an object.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket containing the object.	required
object_name	str	Object name whose tags to retrieve.	required

Returns:

Type	Description
dict[str, str]	dict[str, str]: Mapping of tag key to tag value. Empty dict if no tags.

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to get tags is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def get_object_tags(self, bucket_name: str, object_name: str) -> dict[str, str]:
    """Return the tags attached to an object.

    Args:
        bucket_name: Bucket containing the object.
        object_name: Object name whose tags to retrieve.

    Returns:
        dict[str, str]: Mapping of tag key to tag value. Empty dict if no tags.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to get tags is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.set_bucket_lifecycle abstractmethod

set_bucket_lifecycle(
    bucket_name: str, rules: list[MinioLifecycleRuleType]
) -> None

Set the lifecycle configuration for a bucket.

Lifecycle rules control automatic expiration (TTL) of objects. Each rule is a dict that maps to the S3 LifecycleRule structure. A minimal expiration rule::

{
    "ID": "expire-after-7-days",
    "Status": "Enabled",
    "Filter": {"Prefix": "uploads/"},
    "Expiration": {"Days": 7},
}

To expire only objects with a specific tag::

{
    "ID": "expire-tagged-7days",
    "Status": "Enabled",
    "Filter": {"Tag": {"Key": "ttl-days", "Value": "7"}},
    "Expiration": {"Days": 7},
}

Parameters:

Name	Type	Description	Default
bucket_name	str	Target bucket name.	required
rules	list[MinioLifecycleRuleType]	List of lifecycle rule dicts (S3 LifecycleRule format).	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty or rules is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to set lifecycle is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def set_bucket_lifecycle(self, bucket_name: str, rules: list[MinioLifecycleRuleType]) -> None:
    """Set the lifecycle configuration for a bucket.

    Lifecycle rules control automatic expiration (TTL) of objects. Each rule is a
    dict that maps to the S3 ``LifecycleRule`` structure. A minimal expiration rule::

        {
            "ID": "expire-after-7-days",
            "Status": "Enabled",
            "Filter": {"Prefix": "uploads/"},
            "Expiration": {"Days": 7},
        }

    To expire only objects with a specific tag::

        {
            "ID": "expire-tagged-7days",
            "Status": "Enabled",
            "Filter": {"Tag": {"Key": "ttl-days", "Value": "7"}},
            "Expiration": {"Days": 7},
        }

    Args:
        bucket_name: Target bucket name.
        rules: List of lifecycle rule dicts (S3 ``LifecycleRule`` format).

    Raises:
        InvalidArgumentError: If bucket_name is empty or rules is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to set lifecycle is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.get_bucket_lifecycle abstractmethod

get_bucket_lifecycle(
    bucket_name: str,
) -> list[MinioLifecycleRuleType]

Return the lifecycle rules configured for a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required

Returns:

Type	Description
list[MinioLifecycleRuleType]	list[MinioLifecycleRuleType]: List of lifecycle rule dicts, or an empty list if no lifecycle configuration is set.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to get lifecycle is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def get_bucket_lifecycle(self, bucket_name: str) -> list[MinioLifecycleRuleType]:
    """Return the lifecycle rules configured for a bucket.

    Args:
        bucket_name: Bucket name.

    Returns:
        list[MinioLifecycleRuleType]: List of lifecycle rule dicts, or an empty
            list if no lifecycle configuration is set.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to get lifecycle is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.delete_bucket_lifecycle abstractmethod

delete_bucket_lifecycle(bucket_name: str) -> None

Delete the lifecycle configuration from a bucket.

After this call the bucket has no lifecycle rules and objects will not be automatically expired.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to delete lifecycle is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def delete_bucket_lifecycle(self, bucket_name: str) -> None:
    """Delete the lifecycle configuration from a bucket.

    After this call the bucket has no lifecycle rules and objects will not be
    automatically expired.

    Args:
        bucket_name: Bucket name.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to delete lifecycle is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.set_bucket_versioning abstractmethod

set_bucket_versioning(
    bucket_name: str, *, enabled: bool
) -> None

Enable or suspend versioning on a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
enabled	bool	If True, enable versioning; if False, suspend it.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to set versioning is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def set_bucket_versioning(self, bucket_name: str, *, enabled: bool) -> None:
    """Enable or suspend versioning on a bucket.

    Args:
        bucket_name: Bucket name.
        enabled: If True, enable versioning; if False, suspend it.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to set versioning is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.get_bucket_versioning abstractmethod

get_bucket_versioning(bucket_name: str) -> str

Return the versioning status of a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required

Returns:

Name	Type	Description
str	str	Versioning status — Enabled, Suspended, or empty string if never configured.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to get versioning is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def get_bucket_versioning(self, bucket_name: str) -> str:
    """Return the versioning status of a bucket.

    Args:
        bucket_name: Bucket name.

    Returns:
        str: Versioning status — ``Enabled``, ``Suspended``, or empty string if never configured.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to get versioning is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.list_object_versions abstractmethod

list_object_versions(
    bucket_name: str, prefix: str = ""
) -> list[MinioObjectVersionType]

List all versions and delete markers for objects in a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
prefix	str	Optional prefix to filter objects by.	''

Returns:

Type	Description
list[MinioObjectVersionType]	list[MinioObjectVersionType]: Version entries with keys, version IDs, and metadata.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to list versions is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def list_object_versions(
    self,
    bucket_name: str,
    prefix: str = "",
) -> list[MinioObjectVersionType]:
    """List all versions and delete markers for objects in a bucket.

    Args:
        bucket_name: Bucket name.
        prefix: Optional prefix to filter objects by.

    Returns:
        list[MinioObjectVersionType]: Version entries with keys, version IDs, and metadata.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to list versions is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

archipy.adapters.minio.ports.MinioPort.remove_object_version abstractmethod

remove_object_version(
    bucket_name: str, object_name: str, version_id: str
) -> None

Permanently delete a specific object version.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object key.	required
version_id	str	Version ID to delete permanently.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket, object, or version does not exist.
PermissionDeniedError	If permission to delete the version is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/ports.py

@abstractmethod
def remove_object_version(self, bucket_name: str, object_name: str, version_id: str) -> None:
    """Permanently delete a specific object version.

    Args:
        bucket_name: Bucket name.
        object_name: Object key.
        version_id: Version ID to delete permanently.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket, object, or version does not exist.
        PermissionDeniedError: If permission to delete the version is denied.
        StorageError: If there's a storage-related error.
    """
    raise NotImplementedError

options: show_root_toc_entry: false heading_level: 3

Adapters

Concrete MinIO adapter wrapping boto3 with ArchiPy conventions for S3-compatible object storage operations.

archipy.adapters.minio.adapters.T module-attribute

T = TypeVar('T')

archipy.adapters.minio.adapters.F module-attribute

F = TypeVar('F', bound=Callable[..., Any])

archipy.adapters.minio.adapters.logger module-attribute

logger = logging.getLogger(__name__)

archipy.adapters.minio.adapters.MinioExceptionHandlerMixin

Mixin class to handle boto3 S3 exceptions in a consistent way.

Source code in archipy/adapters/minio/adapters.py

class MinioExceptionHandlerMixin:
    """Mixin class to handle boto3 S3 exceptions in a consistent way."""

    @classmethod
    def _handle_client_exception(cls, exception: ClientError, operation: str) -> NoReturn:
        """Handle ClientError exceptions and map them to appropriate application errors.

        Args:
            exception: The original ClientError exception from boto3
            operation: The name of the operation that failed

        Raises:
            Various application-specific errors based on the exception error code
        """
        error_code = exception.response.get("Error", {}).get("Code", "")

        # Bucket existence errors
        if error_code in ("NoSuchBucket", "404"):
            raise NotFoundError(resource_type="bucket") from exception

        # Object existence errors
        if error_code in ("NoSuchKey", "NoSuchObject"):
            raise NotFoundError(resource_type="object") from exception

        # Bucket ownership/existence errors
        if error_code in ("BucketAlreadyOwnedByYou", "BucketAlreadyExists"):
            raise AlreadyExistsError(resource_type="bucket") from exception

        # Permission errors
        if error_code in ("AccessDenied", "Forbidden", "403"):
            raise PermissionDeniedError(
                additional_data={"details": f"Permission denied for operation: {operation}"},
            ) from exception

        # Resource limit errors
        if error_code in ("QuotaExceeded", "LimitExceeded", "TooManyBuckets"):
            raise ResourceExhaustedError(resource_type="storage") from exception

        # Invalid parameter errors
        if error_code in ("InvalidArgument", "InvalidParameterValue", "InvalidRequest"):
            raise InvalidArgumentError(argument_name=operation) from exception

        # Default: general storage error
        raise StorageError(additional_data={"operation": operation, "error_code": error_code}) from exception

    @classmethod
    def _handle_connection_exception(cls, exception: Exception, _operation: str) -> NoReturn:
        """Handle connection-related exceptions.

        Args:
            exception: The original exception
            operation: The operation that failed

        Raises:
            ConnectionTimeoutError or NetworkError or ServiceUnavailableError
        """
        error_msg = str(exception).lower()

        if "timeout" in error_msg or "timed out" in error_msg:
            raise ConnectionTimeoutError(service="S3") from exception

        if isinstance(exception, (BotocoreConnectionError, EndpointConnectionError)):
            raise NetworkError(service="S3") from exception

        if "unavailable" in error_msg or "connection" in error_msg:
            raise ServiceUnavailableError(service="S3") from exception

        raise NetworkError(service="S3") from exception

    @classmethod
    def _handle_general_exception(cls, exception: Exception, component: str) -> NoReturn:
        """Handle general exceptions by converting them to appropriate application errors.

        Args:
            exception: The original exception
            component: The component/operation name for context

        Raises:
            InternalError: A wrapped version of the original exception
        """
        raise InternalError(additional_data={"component": component}) from exception

archipy.adapters.minio.adapters.MinioAdapter

Bases: MinioPort, MinioExceptionHandlerMixin

Concrete implementation of the MinioPort interface using boto3.

Source code in archipy/adapters/minio/adapters.py

class MinioAdapter(MinioPort, MinioExceptionHandlerMixin):
    """Concrete implementation of the MinioPort interface using boto3."""

    def __init__(self, minio_configs: MinioConfig | None = None) -> None:
        """Initialize MinioAdapter with configuration.

        Args:
            minio_configs: Optional MinIO/S3 configuration. If None, global config is used.

        Raises:
            ConfigurationError: If there is an error in the configuration.
            InvalidArgumentError: If required parameters are missing.
            NetworkError: If there are network errors connecting to S3/MinIO server.
        """
        try:
            # Determine config source (explicit or from global config)
            if minio_configs is not None:
                self.configs = minio_configs
            else:
                # First get global config, then extract MINIO config
                global_config = BaseConfig.global_config()
                if not hasattr(global_config, "MINIO"):
                    raise InvalidArgumentError(argument_name="MINIO")
                minio_config = getattr(global_config, "MINIO", None)
                if not isinstance(minio_config, MinioConfig):
                    raise InvalidArgumentError(argument_name="MINIO")
                self.configs = minio_config

            # Ensure we have a valid endpoint value
            endpoint = str(self.configs.ENDPOINT or "")
            if not endpoint:
                raise InvalidArgumentError(argument_name="endpoint")

            # Determine SSL usage (USE_SSL overrides SECURE if set)
            use_ssl = self.configs.USE_SSL if self.configs.USE_SSL is not None else self.configs.SECURE

            # Construct endpoint URL with protocol
            protocol = "https" if use_ssl else "http"
            endpoint_url = f"{protocol}://{endpoint}"

            # Configure boto3 client with retry and timeout settings
            boto_config = Config(
                signature_version=self.configs.SIGNATURE_VERSION,
                s3={
                    "addressing_style": self.configs.ADDRESSING_STYLE,
                },
                retries={
                    "max_attempts": self.configs.RETRIES_MAX_ATTEMPTS,
                    "mode": self.configs.RETRIES_MODE,
                },
                connect_timeout=self.configs.CONNECT_TIMEOUT,
                read_timeout=self.configs.READ_TIMEOUT,
                max_pool_connections=self.configs.MAX_POOL_CONNECTIONS,
            )

            # Create boto3 S3 client
            self._client = boto3.client(
                "s3",
                endpoint_url=endpoint_url,
                aws_access_key_id=self.configs.ACCESS_KEY,
                aws_secret_access_key=self.configs.SECRET_KEY,
                aws_session_token=self.configs.SESSION_TOKEN,
                region_name=self.configs.REGION,
                config=boto_config,
                verify=self.configs.VERIFY_SSL,
            )
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            error_code = e.response.get("Error", {}).get("Code", "")
            if "configuration" in error_code.lower():
                raise ConfigurationError(operation="s3") from e
            else:
                raise InternalError(additional_data={"component": "S3"}) from e
        except (BotocoreConnectionError, EndpointConnectionError) as e:
            raise NetworkError(service="S3") from e
        except Exception as e:
            raise InternalError(additional_data={"component": "S3"}) from e

    def clear_all_caches(self) -> None:
        """Clear all cached values."""
        for attr_name in dir(self):
            attr = getattr(self, attr_name)
            if hasattr(attr, "clear_cache"):
                attr.clear_cache()

    def _clear_object_caches(self) -> None:
        """Clear caches for object listing and existence checks."""
        if hasattr(self.list_objects, "clear_cache"):
            self.list_objects.clear_cache()
        if hasattr(self.object_exists, "clear_cache"):
            self.object_exists.clear_cache()

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)
    def bucket_exists(self, bucket_name: str) -> bool:
        """Check if a bucket exists.

        Args:
            bucket_name: Name of the bucket to check.

        Returns:
            bool: True if bucket exists, False otherwise.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            self._client.head_bucket(Bucket=bucket_name)
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            error_code = e.response.get("Error", {}).get("Code", "")
            if error_code in ("NoSuchBucket", "404"):
                return False
            self._handle_client_exception(e, "bucket_exists")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "bucket_exists")
            raise
        except Exception as e:
            self._handle_general_exception(e, "bucket_exists")
            raise
        else:
            return True

    @override
    def make_bucket(self, bucket_name: str) -> None:
        """Create a new bucket.

        Args:
            bucket_name: Name of the bucket to create.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            AlreadyExistsError: If the bucket already exists.
            PermissionDeniedError: If permission to create bucket is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")

            # Handle region-specific bucket creation
            create_bucket_config = {}
            if self.configs.REGION and self.configs.REGION != "us-east-1":
                create_bucket_config = {"CreateBucketConfiguration": {"LocationConstraint": self.configs.REGION}}

            if create_bucket_config:
                self._client.create_bucket(Bucket=bucket_name, **create_bucket_config)
            else:
                self._client.create_bucket(Bucket=bucket_name)

            self.clear_all_caches()
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "make_bucket")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "make_bucket")
        except Exception as e:
            self._handle_general_exception(e, "make_bucket")

    @override
    def remove_bucket(self, bucket_name: str) -> None:
        """Remove a bucket.

        Args:
            bucket_name: Name of the bucket to remove.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to delete bucket is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            self._client.delete_bucket(Bucket=bucket_name)
            self.clear_all_caches()
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "remove_bucket")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "remove_bucket")
        except Exception as e:
            self._handle_general_exception(e, "remove_bucket")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=1)
    def list_buckets(self) -> list[MinioBucketType]:
        """List all buckets.

        Returns:
            list: List of buckets and their creation dates.

        Raises:
            PermissionDeniedError: If permission to list buckets is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            response = self._client.list_buckets()
        except ClientError as e:
            self._handle_client_exception(e, "list_buckets")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "list_buckets")
            raise
        except Exception as e:
            self._handle_general_exception(e, "list_buckets")
            raise
        else:
            # Convert buckets to MinioBucketType format
            buckets = response.get("Buckets", [])
            bucket_list: list[MinioBucketType] = [
                {"name": b["Name"], "creation_date": b["CreationDate"]} for b in buckets
            ]
            return bucket_list

    @override
    def put_object(
        self,
        bucket_name: str,
        object_name: str,
        file_path: str,
        tags: dict[str, str] | None = None,
    ) -> None:
        """Upload a file to a bucket.

        Args:
            bucket_name: Destination bucket name.
            object_name: Object name in the bucket.
            file_path: Local file path to upload.
            tags: Optional key-value tags to attach to the object. Tags can be used
                with bucket lifecycle rules to implement per-object TTL expiration.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to upload is denied.
            ResourceExhaustedError: If storage limits are exceeded.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name or not file_path:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name, object_name or file_path"
                        if not all([bucket_name, object_name, file_path])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                        if not object_name
                        else "file_path"
                    ),
                )
            extra_args: dict[str, Any] = {}
            if tags:
                extra_args["Tagging"] = urllib.parse.urlencode(tags)
            self._client.upload_file(file_path, bucket_name, object_name, ExtraArgs=extra_args or None)
            self._clear_object_caches()
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "put_object")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "put_object")
        except Exception as e:
            self._handle_general_exception(e, "put_object")

    @override
    def get_object(self, bucket_name: str, object_name: str, file_path: str) -> None:
        """Download an object to a file.

        Args:
            bucket_name: Source bucket name.
            object_name: Object name in the bucket.
            file_path: Local file path to save the object.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to download is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name or not file_path:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name, object_name or file_path"
                        if not all([bucket_name, object_name, file_path])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                        if not object_name
                        else "file_path"
                    ),
                )
            self._client.download_file(bucket_name, object_name, file_path)
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "get_object")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "get_object")
        except Exception as e:
            self._handle_general_exception(e, "get_object")

    @override
    def remove_object(self, bucket_name: str, object_name: str) -> None:
        """Remove an object from a bucket.

        Args:
            bucket_name: Bucket name.
            object_name: Object name to remove.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to remove is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            self._client.delete_object(Bucket=bucket_name, Key=object_name)
            self._clear_object_caches()
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "remove_object")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "remove_object")
        except Exception as e:
            self._handle_general_exception(e, "remove_object")

    @override
    def remove_objects(self, bucket_name: str, object_names: list[str]) -> None:
        """Remove multiple objects from a bucket in a single request.

        Args:
            bucket_name: Bucket name.
            object_names: Object keys to delete.

        Raises:
            InvalidArgumentError: If bucket_name is empty or object_names is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to remove objects is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            if not object_names:
                raise InvalidArgumentError(argument_name="object_names")

            delete_payload = {"Objects": [{"Key": name} for name in object_names], "Quiet": True}
            self._client.delete_objects(Bucket=bucket_name, Delete=delete_payload)
            self._clear_object_caches()
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "remove_objects")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "remove_objects")
        except Exception as e:
            self._handle_general_exception(e, "remove_objects")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)
    def object_exists(self, bucket_name: str, object_name: str) -> bool:
        """Check if an object exists in a bucket.

        Args:
            bucket_name: Bucket name.
            object_name: Object key to check.

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

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            PermissionDeniedError: If permission to check the object is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            self._client.head_object(Bucket=bucket_name, Key=object_name)
        except InvalidArgumentError:
            raise
        except ClientError as e:
            error_code = e.response.get("Error", {}).get("Code", "")
            if error_code in ("NoSuchKey", "NoSuchObject", "404"):
                return False
            self._handle_client_exception(e, "object_exists")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "object_exists")
            raise
        except Exception as e:
            self._handle_general_exception(e, "object_exists")
            raise
        else:
            return True

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)
    def list_objects(
        self,
        bucket_name: str,
        prefix: str = "",
        *,
        recursive: bool = False,
    ) -> list[MinioObjectType]:
        """List objects in a bucket.

        Args:
            bucket_name: Bucket name.
            prefix: Optional prefix to filter objects.
            recursive: Whether to list objects recursively.

        Returns:
            list: List of objects with metadata.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to list objects is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")

            # Build list_objects_v2 parameters
            params = {"Bucket": bucket_name}
            if prefix:
                params["Prefix"] = prefix
            if not recursive:
                params["Delimiter"] = "/"

            # Handle pagination
            object_list: list[MinioObjectType] = []
            paginator = self._client.get_paginator("list_objects_v2")
            for page in paginator.paginate(**params):
                object_list.extend(
                    {
                        "object_name": obj["Key"],
                        "size": obj["Size"],
                        "last_modified": obj["LastModified"],
                    }
                    for obj in page.get("Contents", [])
                )
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "list_objects")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "list_objects")
            raise
        except Exception as e:
            self._handle_general_exception(e, "list_objects")
            raise
        else:
            return object_list

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)
    def stat_object(self, bucket_name: str, object_name: str) -> MinioObjectType:
        """Get object metadata.

        Args:
            bucket_name: Bucket name.
            object_name: Object name to get stats for.

        Returns:
            dict: Object metadata including name, size, last modified date, etc.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to get stats is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            response = self._client.head_object(Bucket=bucket_name, Key=object_name)
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "stat_object")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "stat_object")
            raise
        except Exception as e:
            self._handle_general_exception(e, "stat_object")
            raise
        else:
            # Convert response to MinioObjectType format
            return {
                "object_name": object_name,
                "size": response.get("ContentLength", 0),
                "last_modified": response.get("LastModified"),
                "content_type": response.get("ContentType"),
                "etag": response.get("ETag", "").strip('"'),
            }

    @override
    def presigned_get_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
        """Generate a presigned URL for downloading an object.

        Args:
            bucket_name: Bucket name.
            object_name: Object name to generate URL for.
            expires: URL expiry time in seconds.

        Returns:
            str: Presigned URL for downloading the object.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to generate URL is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            url = self._client.generate_presigned_url(
                "get_object",
                Params={"Bucket": bucket_name, "Key": object_name},
                ExpiresIn=expires,
            )
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "presigned_get_object")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "presigned_get_object")
            raise
        except Exception as e:
            self._handle_general_exception(e, "presigned_get_object")
            raise
        else:
            typed_url: str = url
            return typed_url

    @override
    def presigned_put_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
        """Generate a presigned URL for uploading an object.

        Args:
            bucket_name: Bucket name.
            object_name: Object name to generate URL for.
            expires: URL expiry time in seconds.

        Returns:
            str: Presigned URL for uploading the object.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to generate URL is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            url = self._client.generate_presigned_url(
                "put_object",
                Params={"Bucket": bucket_name, "Key": object_name},
                ExpiresIn=expires,
            )
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "presigned_put_object")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "presigned_put_object")
            raise
        except Exception as e:
            self._handle_general_exception(e, "presigned_put_object")
            raise
        else:
            typed_url: str = url
            return typed_url

    @override
    def set_bucket_policy(self, bucket_name: str, policy: str) -> None:
        """Set bucket policy.

        Args:
            bucket_name: Bucket name.
            policy: JSON policy string.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to set policy is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not policy:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or policy"
                        if not all([bucket_name, policy])
                        else "bucket_name"
                        if not bucket_name
                        else "policy"
                    ),
                )
            self._client.put_bucket_policy(Bucket=bucket_name, Policy=policy)
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "set_bucket_policy")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "set_bucket_policy")
        except Exception as e:
            self._handle_general_exception(e, "set_bucket_policy")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)
    def get_bucket_policy(self, bucket_name: str) -> MinioPolicyType:
        """Get bucket policy.

        Args:
            bucket_name: Bucket name.

        Returns:
            dict: Bucket policy information.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to get policy is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            response = self._client.get_bucket_policy(Bucket=bucket_name)
            policy = response.get("Policy", "{}")
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "get_bucket_policy")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "get_bucket_policy")
            raise
        except Exception as e:
            self._handle_general_exception(e, "get_bucket_policy")
            raise
        else:
            # Convert policy to MinioPolicyType format
            policy_dict: MinioPolicyType = {"policy": policy}
            return policy_dict

    @override
    def copy_object(
        self,
        src_bucket_name: str,
        src_object_name: str,
        dest_bucket_name: str,
        dest_object_name: str,
    ) -> None:
        """Copy an object within or between buckets.

        Args:
            src_bucket_name: Source bucket name.
            src_object_name: Source object name.
            dest_bucket_name: Destination bucket name.
            dest_object_name: Destination object name.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the source bucket or object does not exist.
            PermissionDeniedError: If permission to copy is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not src_bucket_name or not src_object_name or not dest_bucket_name or not dest_object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "src_bucket_name, src_object_name, dest_bucket_name or dest_object_name"
                        if not all([src_bucket_name, src_object_name, dest_bucket_name, dest_object_name])
                        else "src_bucket_name"
                        if not src_bucket_name
                        else "src_object_name"
                        if not src_object_name
                        else "dest_bucket_name"
                        if not dest_bucket_name
                        else "dest_object_name"
                    ),
                )
            self._client.copy_object(
                Bucket=dest_bucket_name,
                CopySource=f"{src_bucket_name}/{src_object_name}",
                Key=dest_object_name,
            )
            self._clear_object_caches()
        except InvalidArgumentError:
            # Pass through our custom errors
            raise
        except ClientError as e:
            self._handle_client_exception(e, "copy_object")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "copy_object")
        except Exception as e:
            self._handle_general_exception(e, "copy_object")

    @override
    def put_object_stream(
        self,
        bucket_name: str,
        object_name: str,
        data: bytes | BinaryIO,
        length: int = -1,
        content_type: str = "application/octet-stream",
        tags: dict[str, str] | None = None,
    ) -> None:
        """Upload data from a bytes buffer or binary stream to a bucket.

        Unlike put_object which requires a local file path, this method accepts
        in-memory bytes or any binary stream, avoiding the need for a temporary file.

        Args:
            bucket_name: Destination bucket name.
            object_name: Object name in the bucket.
            data: Content to upload as raw bytes or a binary stream (BinaryIO).
            length: Content length in bytes. If -1, computed automatically for bytes;
                for streams, providing the exact length avoids buffering overhead.
            content_type: MIME type of the content. Defaults to "application/octet-stream".
            tags: Optional key-value tags to attach to the object. Tags can be used
                with bucket lifecycle rules to implement per-object TTL expiration.

        Raises:
            InvalidArgumentError: If bucket_name, object_name, or data is invalid.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to upload is denied.
            ResourceExhaustedError: If storage limits are exceeded.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )

            kwargs: dict[str, Any] = {
                "Bucket": bucket_name,
                "Key": object_name,
                "Body": data,
                "ContentType": content_type,
            }

            if isinstance(data, bytes):
                kwargs["ContentLength"] = len(data) if length < 0 else length
            elif length >= 0:
                kwargs["ContentLength"] = length

            if tags:
                kwargs["Tagging"] = urllib.parse.urlencode(tags)

            self._client.put_object(**kwargs)
            self._clear_object_caches()
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "put_object_stream")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "put_object_stream")
        except Exception as e:
            self._handle_general_exception(e, "put_object_stream")

    @override
    def get_object_stream(self, bucket_name: str, object_name: str) -> bytes:
        """Download an object and return its content as bytes.

        Unlike get_object which requires a local file path, this method returns
        the object content directly in memory, avoiding a temporary file.

        Args:
            bucket_name: Source bucket name.
            object_name: Object name in the bucket.

        Returns:
            bytes: The full content of the object.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to download is denied.
            ServiceUnavailableError: If the S3 service is unavailable.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            response = self._client.get_object(Bucket=bucket_name, Key=object_name)
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "get_object_stream")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "get_object_stream")
            raise
        except Exception as e:
            self._handle_general_exception(e, "get_object_stream")
            raise
        else:
            content: bytes = response["Body"].read()
            return content

    @override
    def set_object_tags(self, bucket_name: str, object_name: str, tags: dict[str, str]) -> None:
        """Set tags on an existing object, replacing any existing tags.

        Tags are key-value pairs that can be used with bucket lifecycle rules to
        implement TTL-based expiration. For example, tag an object with
        ``{"ttl-days": "7"}`` and create a matching lifecycle rule to auto-delete
        it after 7 days.

        Args:
            bucket_name: Bucket containing the object.
            object_name: Object name to tag.
            tags: Key-value mapping of tag names to values.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to set tags is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            tag_set = [{"Key": k, "Value": v} for k, v in tags.items()]
            self._client.put_object_tagging(
                Bucket=bucket_name,
                Key=object_name,
                Tagging={"TagSet": tag_set},
            )
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "set_object_tags")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "set_object_tags")
        except Exception as e:
            self._handle_general_exception(e, "set_object_tags")

    @override
    def remove_object_tags(self, bucket_name: str, object_name: str) -> None:
        """Remove all tags from an object.

        Args:
            bucket_name: Bucket containing the object.
            object_name: Object name whose tags to remove.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to remove tags is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            self._client.delete_object_tagging(Bucket=bucket_name, Key=object_name)
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "remove_object_tags")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "remove_object_tags")
        except Exception as e:
            self._handle_general_exception(e, "remove_object_tags")

    @override
    def get_object_tags(self, bucket_name: str, object_name: str) -> dict[str, str]:
        """Return the tags attached to an object.

        Args:
            bucket_name: Bucket containing the object.
            object_name: Object name whose tags to retrieve.

        Returns:
            dict[str, str]: Mapping of tag key to tag value. Empty dict if no tags.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket or object does not exist.
            PermissionDeniedError: If permission to get tags is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name or object_name"
                        if not all([bucket_name, object_name])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                    ),
                )
            response = self._client.get_object_tagging(Bucket=bucket_name, Key=object_name)
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "get_object_tags")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "get_object_tags")
            raise
        except Exception as e:
            self._handle_general_exception(e, "get_object_tags")
            raise
        else:
            return {tag["Key"]: tag["Value"] for tag in response.get("TagSet", [])}

    @override
    def set_bucket_lifecycle(self, bucket_name: str, rules: list[MinioLifecycleRuleType]) -> None:
        """Set the lifecycle configuration for a bucket.

        Lifecycle rules control automatic expiration (TTL) of objects. Each rule is a
        dict that maps to the S3 ``LifecycleRule`` structure. A minimal expiration rule::

            {
                "ID": "expire-after-7-days",
                "Status": "Enabled",
                "Filter": {"Prefix": "uploads/"},
                "Expiration": {"Days": 7},
            }

        To expire only objects with a specific tag::

            {
                "ID": "expire-tagged-7days",
                "Status": "Enabled",
                "Filter": {"Tag": {"Key": "ttl-days", "Value": "7"}},
                "Expiration": {"Days": 7},
            }

        Args:
            bucket_name: Target bucket name.
            rules: List of lifecycle rule dicts (S3 ``LifecycleRule`` format).

        Raises:
            InvalidArgumentError: If bucket_name is empty or rules is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to set lifecycle is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            if not rules:
                raise InvalidArgumentError(argument_name="rules")
            self._client.put_bucket_lifecycle_configuration(
                Bucket=bucket_name,
                LifecycleConfiguration={"Rules": rules},
            )
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "set_bucket_lifecycle")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "set_bucket_lifecycle")
        except Exception as e:
            self._handle_general_exception(e, "set_bucket_lifecycle")

    @override
    def get_bucket_lifecycle(self, bucket_name: str) -> list[MinioLifecycleRuleType]:
        """Return the lifecycle rules configured for a bucket.

        Args:
            bucket_name: Bucket name.

        Returns:
            list[MinioLifecycleRuleType]: List of lifecycle rule dicts, or an empty
                list if no lifecycle configuration is set.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to get lifecycle is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            response = self._client.get_bucket_lifecycle_configuration(Bucket=bucket_name)
        except InvalidArgumentError:
            raise
        except ClientError as e:
            if e.response.get("Error", {}).get("Code", "") == "NoSuchLifecycleConfiguration":
                return []
            self._handle_client_exception(e, "get_bucket_lifecycle")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "get_bucket_lifecycle")
            raise
        except Exception as e:
            self._handle_general_exception(e, "get_bucket_lifecycle")
            raise
        else:
            rules: list[MinioLifecycleRuleType] = response.get("Rules", [])
            return rules

    @override
    def delete_bucket_lifecycle(self, bucket_name: str) -> None:
        """Delete the lifecycle configuration from a bucket.

        After this call the bucket has no lifecycle rules and objects will not be
        automatically expired.

        Args:
            bucket_name: Bucket name.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to delete lifecycle is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            self._client.delete_bucket_lifecycle(Bucket=bucket_name)
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "delete_bucket_lifecycle")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "delete_bucket_lifecycle")
        except Exception as e:
            self._handle_general_exception(e, "delete_bucket_lifecycle")

    @override
    def set_bucket_versioning(self, bucket_name: str, *, enabled: bool) -> None:
        """Enable or suspend versioning on a bucket.

        Args:
            bucket_name: Bucket name.
            enabled: If True, enable versioning; if False, suspend it.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to set versioning is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            status = "Enabled" if enabled else "Suspended"
            self._client.put_bucket_versioning(
                Bucket=bucket_name,
                VersioningConfiguration={"Status": status},
            )
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "set_bucket_versioning")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "set_bucket_versioning")
        except Exception as e:
            self._handle_general_exception(e, "set_bucket_versioning")

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)
    def get_bucket_versioning(self, bucket_name: str) -> str:
        """Return the versioning status of a bucket.

        Args:
            bucket_name: Bucket name.

        Returns:
            str: Versioning status — ``Enabled``, ``Suspended``, or empty string if never configured.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to get versioning is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")
            response = self._client.get_bucket_versioning(Bucket=bucket_name)
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "get_bucket_versioning")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "get_bucket_versioning")
            raise
        except Exception as e:
            self._handle_general_exception(e, "get_bucket_versioning")
            raise
        else:
            status: str = response.get("Status", "")
            return status

    @override
    @ttl_cache_decorator(ttl_seconds=300, maxsize=100)
    def list_object_versions(
        self,
        bucket_name: str,
        prefix: str = "",
    ) -> list[MinioObjectVersionType]:
        """List all versions and delete markers for objects in a bucket.

        Args:
            bucket_name: Bucket name.
            prefix: Optional prefix to filter objects by.

        Returns:
            list[MinioObjectVersionType]: Version entries with keys, version IDs, and metadata.

        Raises:
            InvalidArgumentError: If bucket_name is empty.
            NotFoundError: If the bucket does not exist.
            PermissionDeniedError: If permission to list versions is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name:
                raise InvalidArgumentError(argument_name="bucket_name")

            params: dict[str, str] = {"Bucket": bucket_name}
            if prefix:
                params["Prefix"] = prefix

            version_list: list[MinioObjectVersionType] = []
            paginator = self._client.get_paginator("list_object_versions")
            for page in paginator.paginate(**params):
                version_list.extend(
                    {
                        "object_name": version["Key"],
                        "version_id": version["VersionId"],
                        "is_latest": version.get("IsLatest", False),
                        "is_delete_marker": False,
                        "size": version.get("Size", 0),
                        "last_modified": version.get("LastModified"),
                    }
                    for version in page.get("Versions", [])
                )
                version_list.extend(
                    {
                        "object_name": marker["Key"],
                        "version_id": marker["VersionId"],
                        "is_latest": marker.get("IsLatest", False),
                        "is_delete_marker": True,
                        "size": 0,
                        "last_modified": marker.get("LastModified"),
                    }
                    for marker in page.get("DeleteMarkers", [])
                )
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "list_object_versions")
            raise
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "list_object_versions")
            raise
        except Exception as e:
            self._handle_general_exception(e, "list_object_versions")
            raise
        else:
            return version_list

    @override
    def remove_object_version(self, bucket_name: str, object_name: str, version_id: str) -> None:
        """Permanently delete a specific object version.

        Args:
            bucket_name: Bucket name.
            object_name: Object key.
            version_id: Version ID to delete permanently.

        Raises:
            InvalidArgumentError: If any required parameter is empty.
            NotFoundError: If the bucket, object, or version does not exist.
            PermissionDeniedError: If permission to delete the version is denied.
            StorageError: If there's a storage-related error.
        """
        try:
            if not bucket_name or not object_name or not version_id:
                raise InvalidArgumentError(
                    argument_name=(
                        "bucket_name, object_name or version_id"
                        if not all([bucket_name, object_name, version_id])
                        else "bucket_name"
                        if not bucket_name
                        else "object_name"
                        if not object_name
                        else "version_id"
                    ),
                )
            self._client.delete_object(Bucket=bucket_name, Key=object_name, VersionId=version_id)
            self._clear_object_caches()
        except InvalidArgumentError:
            raise
        except ClientError as e:
            self._handle_client_exception(e, "remove_object_version")
        except (ConnectionError, EndpointConnectionError) as e:
            self._handle_connection_exception(e, "remove_object_version")
        except Exception as e:
            self._handle_general_exception(e, "remove_object_version")

archipy.adapters.minio.adapters.MinioAdapter.configs instance-attribute

configs = minio_configs

archipy.adapters.minio.adapters.MinioAdapter.clear_all_caches

clear_all_caches() -> None

Clear all cached values.

Source code in archipy/adapters/minio/adapters.py

def clear_all_caches(self) -> None:
    """Clear all cached values."""
    for attr_name in dir(self):
        attr = getattr(self, attr_name)
        if hasattr(attr, "clear_cache"):
            attr.clear_cache()

archipy.adapters.minio.adapters.MinioAdapter.bucket_exists

bucket_exists(bucket_name: str) -> bool

Check if a bucket exists.

Parameters:

Name	Type	Description	Default
bucket_name	str	Name of the bucket to check.	required

Returns:

Name	Type	Description
bool	bool	True if bucket exists, False otherwise.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)
def bucket_exists(self, bucket_name: str) -> bool:
    """Check if a bucket exists.

    Args:
        bucket_name: Name of the bucket to check.

    Returns:
        bool: True if bucket exists, False otherwise.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        self._client.head_bucket(Bucket=bucket_name)
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        error_code = e.response.get("Error", {}).get("Code", "")
        if error_code in ("NoSuchBucket", "404"):
            return False
        self._handle_client_exception(e, "bucket_exists")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "bucket_exists")
        raise
    except Exception as e:
        self._handle_general_exception(e, "bucket_exists")
        raise
    else:
        return True

archipy.adapters.minio.adapters.MinioAdapter.make_bucket

make_bucket(bucket_name: str) -> None

Create a new bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Name of the bucket to create.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
AlreadyExistsError	If the bucket already exists.
PermissionDeniedError	If permission to create bucket is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def make_bucket(self, bucket_name: str) -> None:
    """Create a new bucket.

    Args:
        bucket_name: Name of the bucket to create.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        AlreadyExistsError: If the bucket already exists.
        PermissionDeniedError: If permission to create bucket is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")

        # Handle region-specific bucket creation
        create_bucket_config = {}
        if self.configs.REGION and self.configs.REGION != "us-east-1":
            create_bucket_config = {"CreateBucketConfiguration": {"LocationConstraint": self.configs.REGION}}

        if create_bucket_config:
            self._client.create_bucket(Bucket=bucket_name, **create_bucket_config)
        else:
            self._client.create_bucket(Bucket=bucket_name)

        self.clear_all_caches()
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "make_bucket")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "make_bucket")
    except Exception as e:
        self._handle_general_exception(e, "make_bucket")

archipy.adapters.minio.adapters.MinioAdapter.remove_bucket

remove_bucket(bucket_name: str) -> None

Remove a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Name of the bucket to remove.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to delete bucket is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def remove_bucket(self, bucket_name: str) -> None:
    """Remove a bucket.

    Args:
        bucket_name: Name of the bucket to remove.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to delete bucket is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        self._client.delete_bucket(Bucket=bucket_name)
        self.clear_all_caches()
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "remove_bucket")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "remove_bucket")
    except Exception as e:
        self._handle_general_exception(e, "remove_bucket")

archipy.adapters.minio.adapters.MinioAdapter.list_buckets

list_buckets() -> list[MinioBucketType]

List all buckets.

Returns:

Name	Type	Description
list	list[MinioBucketType]	List of buckets and their creation dates.

Raises:

Type	Description
PermissionDeniedError	If permission to list buckets is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=1)
def list_buckets(self) -> list[MinioBucketType]:
    """List all buckets.

    Returns:
        list: List of buckets and their creation dates.

    Raises:
        PermissionDeniedError: If permission to list buckets is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        response = self._client.list_buckets()
    except ClientError as e:
        self._handle_client_exception(e, "list_buckets")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "list_buckets")
        raise
    except Exception as e:
        self._handle_general_exception(e, "list_buckets")
        raise
    else:
        # Convert buckets to MinioBucketType format
        buckets = response.get("Buckets", [])
        bucket_list: list[MinioBucketType] = [
            {"name": b["Name"], "creation_date": b["CreationDate"]} for b in buckets
        ]
        return bucket_list

archipy.adapters.minio.adapters.MinioAdapter.put_object

put_object(
    bucket_name: str,
    object_name: str,
    file_path: str,
    tags: dict[str, str] | None = None,
) -> None

Upload a file to a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Destination bucket name.	required
object_name	str	Object name in the bucket.	required
file_path	str	Local file path to upload.	required
tags	dict[str, str] | None	Optional key-value tags to attach to the object. Tags can be used with bucket lifecycle rules to implement per-object TTL expiration.	None

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to upload is denied.
ResourceExhaustedError	If storage limits are exceeded.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def put_object(
    self,
    bucket_name: str,
    object_name: str,
    file_path: str,
    tags: dict[str, str] | None = None,
) -> None:
    """Upload a file to a bucket.

    Args:
        bucket_name: Destination bucket name.
        object_name: Object name in the bucket.
        file_path: Local file path to upload.
        tags: Optional key-value tags to attach to the object. Tags can be used
            with bucket lifecycle rules to implement per-object TTL expiration.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to upload is denied.
        ResourceExhaustedError: If storage limits are exceeded.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name or not file_path:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name, object_name or file_path"
                    if not all([bucket_name, object_name, file_path])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                    if not object_name
                    else "file_path"
                ),
            )
        extra_args: dict[str, Any] = {}
        if tags:
            extra_args["Tagging"] = urllib.parse.urlencode(tags)
        self._client.upload_file(file_path, bucket_name, object_name, ExtraArgs=extra_args or None)
        self._clear_object_caches()
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "put_object")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "put_object")
    except Exception as e:
        self._handle_general_exception(e, "put_object")

archipy.adapters.minio.adapters.MinioAdapter.get_object

get_object(
    bucket_name: str, object_name: str, file_path: str
) -> None

Download an object to a file.

Parameters:

Name	Type	Description	Default
bucket_name	str	Source bucket name.	required
object_name	str	Object name in the bucket.	required
file_path	str	Local file path to save the object.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to download is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def get_object(self, bucket_name: str, object_name: str, file_path: str) -> None:
    """Download an object to a file.

    Args:
        bucket_name: Source bucket name.
        object_name: Object name in the bucket.
        file_path: Local file path to save the object.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to download is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name or not file_path:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name, object_name or file_path"
                    if not all([bucket_name, object_name, file_path])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                    if not object_name
                    else "file_path"
                ),
            )
        self._client.download_file(bucket_name, object_name, file_path)
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "get_object")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "get_object")
    except Exception as e:
        self._handle_general_exception(e, "get_object")

archipy.adapters.minio.adapters.MinioAdapter.remove_object

remove_object(bucket_name: str, object_name: str) -> None

Remove an object from a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object name to remove.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to remove is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def remove_object(self, bucket_name: str, object_name: str) -> None:
    """Remove an object from a bucket.

    Args:
        bucket_name: Bucket name.
        object_name: Object name to remove.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to remove is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        self._client.delete_object(Bucket=bucket_name, Key=object_name)
        self._clear_object_caches()
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "remove_object")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "remove_object")
    except Exception as e:
        self._handle_general_exception(e, "remove_object")

archipy.adapters.minio.adapters.MinioAdapter.remove_objects

remove_objects(
    bucket_name: str, object_names: list[str]
) -> None

Remove multiple objects from a bucket in a single request.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_names	list[str]	Object keys to delete.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty or object_names is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to remove objects is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def remove_objects(self, bucket_name: str, object_names: list[str]) -> None:
    """Remove multiple objects from a bucket in a single request.

    Args:
        bucket_name: Bucket name.
        object_names: Object keys to delete.

    Raises:
        InvalidArgumentError: If bucket_name is empty or object_names is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to remove objects is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        if not object_names:
            raise InvalidArgumentError(argument_name="object_names")

        delete_payload = {"Objects": [{"Key": name} for name in object_names], "Quiet": True}
        self._client.delete_objects(Bucket=bucket_name, Delete=delete_payload)
        self._clear_object_caches()
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "remove_objects")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "remove_objects")
    except Exception as e:
        self._handle_general_exception(e, "remove_objects")

archipy.adapters.minio.adapters.MinioAdapter.object_exists

object_exists(bucket_name: str, object_name: str) -> bool

Check if an object exists in a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object key to check.	required

Returns:

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

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
PermissionDeniedError	If permission to check the object is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)
def object_exists(self, bucket_name: str, object_name: str) -> bool:
    """Check if an object exists in a bucket.

    Args:
        bucket_name: Bucket name.
        object_name: Object key to check.

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

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        PermissionDeniedError: If permission to check the object is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        self._client.head_object(Bucket=bucket_name, Key=object_name)
    except InvalidArgumentError:
        raise
    except ClientError as e:
        error_code = e.response.get("Error", {}).get("Code", "")
        if error_code in ("NoSuchKey", "NoSuchObject", "404"):
            return False
        self._handle_client_exception(e, "object_exists")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "object_exists")
        raise
    except Exception as e:
        self._handle_general_exception(e, "object_exists")
        raise
    else:
        return True

archipy.adapters.minio.adapters.MinioAdapter.list_objects

list_objects(
    bucket_name: str,
    prefix: str = "",
    *,
    recursive: bool = False,
) -> list[MinioObjectType]

List objects in a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
prefix	str	Optional prefix to filter objects.	''
recursive	bool	Whether to list objects recursively.	False

Returns:

Name	Type	Description
list	list[MinioObjectType]	List of objects with metadata.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to list objects is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)
def list_objects(
    self,
    bucket_name: str,
    prefix: str = "",
    *,
    recursive: bool = False,
) -> list[MinioObjectType]:
    """List objects in a bucket.

    Args:
        bucket_name: Bucket name.
        prefix: Optional prefix to filter objects.
        recursive: Whether to list objects recursively.

    Returns:
        list: List of objects with metadata.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to list objects is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")

        # Build list_objects_v2 parameters
        params = {"Bucket": bucket_name}
        if prefix:
            params["Prefix"] = prefix
        if not recursive:
            params["Delimiter"] = "/"

        # Handle pagination
        object_list: list[MinioObjectType] = []
        paginator = self._client.get_paginator("list_objects_v2")
        for page in paginator.paginate(**params):
            object_list.extend(
                {
                    "object_name": obj["Key"],
                    "size": obj["Size"],
                    "last_modified": obj["LastModified"],
                }
                for obj in page.get("Contents", [])
            )
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "list_objects")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "list_objects")
        raise
    except Exception as e:
        self._handle_general_exception(e, "list_objects")
        raise
    else:
        return object_list

archipy.adapters.minio.adapters.MinioAdapter.stat_object

stat_object(
    bucket_name: str, object_name: str
) -> MinioObjectType

Get object metadata.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object name to get stats for.	required

Returns:

Name	Type	Description
dict	MinioObjectType	Object metadata including name, size, last modified date, etc.

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to get stats is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)
def stat_object(self, bucket_name: str, object_name: str) -> MinioObjectType:
    """Get object metadata.

    Args:
        bucket_name: Bucket name.
        object_name: Object name to get stats for.

    Returns:
        dict: Object metadata including name, size, last modified date, etc.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to get stats is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        response = self._client.head_object(Bucket=bucket_name, Key=object_name)
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "stat_object")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "stat_object")
        raise
    except Exception as e:
        self._handle_general_exception(e, "stat_object")
        raise
    else:
        # Convert response to MinioObjectType format
        return {
            "object_name": object_name,
            "size": response.get("ContentLength", 0),
            "last_modified": response.get("LastModified"),
            "content_type": response.get("ContentType"),
            "etag": response.get("ETag", "").strip('"'),
        }

archipy.adapters.minio.adapters.MinioAdapter.presigned_get_object

presigned_get_object(
    bucket_name: str, object_name: str, expires: int = 3600
) -> str

Generate a presigned URL for downloading an object.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object name to generate URL for.	required
expires	int	URL expiry time in seconds.	3600

Returns:

Name	Type	Description
str	str	Presigned URL for downloading the object.

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to generate URL is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def presigned_get_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
    """Generate a presigned URL for downloading an object.

    Args:
        bucket_name: Bucket name.
        object_name: Object name to generate URL for.
        expires: URL expiry time in seconds.

    Returns:
        str: Presigned URL for downloading the object.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to generate URL is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        url = self._client.generate_presigned_url(
            "get_object",
            Params={"Bucket": bucket_name, "Key": object_name},
            ExpiresIn=expires,
        )
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "presigned_get_object")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "presigned_get_object")
        raise
    except Exception as e:
        self._handle_general_exception(e, "presigned_get_object")
        raise
    else:
        typed_url: str = url
        return typed_url

archipy.adapters.minio.adapters.MinioAdapter.presigned_put_object

presigned_put_object(
    bucket_name: str, object_name: str, expires: int = 3600
) -> str

Generate a presigned URL for uploading an object.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object name to generate URL for.	required
expires	int	URL expiry time in seconds.	3600

Returns:

Name	Type	Description
str	str	Presigned URL for uploading the object.

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to generate URL is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def presigned_put_object(self, bucket_name: str, object_name: str, expires: int = 3600) -> str:
    """Generate a presigned URL for uploading an object.

    Args:
        bucket_name: Bucket name.
        object_name: Object name to generate URL for.
        expires: URL expiry time in seconds.

    Returns:
        str: Presigned URL for uploading the object.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to generate URL is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        url = self._client.generate_presigned_url(
            "put_object",
            Params={"Bucket": bucket_name, "Key": object_name},
            ExpiresIn=expires,
        )
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "presigned_put_object")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "presigned_put_object")
        raise
    except Exception as e:
        self._handle_general_exception(e, "presigned_put_object")
        raise
    else:
        typed_url: str = url
        return typed_url

archipy.adapters.minio.adapters.MinioAdapter.set_bucket_policy

set_bucket_policy(bucket_name: str, policy: str) -> None

Set bucket policy.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
policy	str	JSON policy string.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to set policy is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def set_bucket_policy(self, bucket_name: str, policy: str) -> None:
    """Set bucket policy.

    Args:
        bucket_name: Bucket name.
        policy: JSON policy string.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to set policy is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not policy:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or policy"
                    if not all([bucket_name, policy])
                    else "bucket_name"
                    if not bucket_name
                    else "policy"
                ),
            )
        self._client.put_bucket_policy(Bucket=bucket_name, Policy=policy)
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "set_bucket_policy")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "set_bucket_policy")
    except Exception as e:
        self._handle_general_exception(e, "set_bucket_policy")

archipy.adapters.minio.adapters.MinioAdapter.get_bucket_policy

get_bucket_policy(bucket_name: str) -> MinioPolicyType

Get bucket policy.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required

Returns:

Name	Type	Description
dict	MinioPolicyType	Bucket policy information.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to get policy is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)
def get_bucket_policy(self, bucket_name: str) -> MinioPolicyType:
    """Get bucket policy.

    Args:
        bucket_name: Bucket name.

    Returns:
        dict: Bucket policy information.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to get policy is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        response = self._client.get_bucket_policy(Bucket=bucket_name)
        policy = response.get("Policy", "{}")
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "get_bucket_policy")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "get_bucket_policy")
        raise
    except Exception as e:
        self._handle_general_exception(e, "get_bucket_policy")
        raise
    else:
        # Convert policy to MinioPolicyType format
        policy_dict: MinioPolicyType = {"policy": policy}
        return policy_dict

archipy.adapters.minio.adapters.MinioAdapter.copy_object

copy_object(
    src_bucket_name: str,
    src_object_name: str,
    dest_bucket_name: str,
    dest_object_name: str,
) -> None

Copy an object within or between buckets.

Parameters:

Name	Type	Description	Default
src_bucket_name	str	Source bucket name.	required
src_object_name	str	Source object name.	required
dest_bucket_name	str	Destination bucket name.	required
dest_object_name	str	Destination object name.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the source bucket or object does not exist.
PermissionDeniedError	If permission to copy is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def copy_object(
    self,
    src_bucket_name: str,
    src_object_name: str,
    dest_bucket_name: str,
    dest_object_name: str,
) -> None:
    """Copy an object within or between buckets.

    Args:
        src_bucket_name: Source bucket name.
        src_object_name: Source object name.
        dest_bucket_name: Destination bucket name.
        dest_object_name: Destination object name.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the source bucket or object does not exist.
        PermissionDeniedError: If permission to copy is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not src_bucket_name or not src_object_name or not dest_bucket_name or not dest_object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "src_bucket_name, src_object_name, dest_bucket_name or dest_object_name"
                    if not all([src_bucket_name, src_object_name, dest_bucket_name, dest_object_name])
                    else "src_bucket_name"
                    if not src_bucket_name
                    else "src_object_name"
                    if not src_object_name
                    else "dest_bucket_name"
                    if not dest_bucket_name
                    else "dest_object_name"
                ),
            )
        self._client.copy_object(
            Bucket=dest_bucket_name,
            CopySource=f"{src_bucket_name}/{src_object_name}",
            Key=dest_object_name,
        )
        self._clear_object_caches()
    except InvalidArgumentError:
        # Pass through our custom errors
        raise
    except ClientError as e:
        self._handle_client_exception(e, "copy_object")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "copy_object")
    except Exception as e:
        self._handle_general_exception(e, "copy_object")

archipy.adapters.minio.adapters.MinioAdapter.put_object_stream

put_object_stream(
    bucket_name: str,
    object_name: str,
    data: bytes | BinaryIO,
    length: int = -1,
    content_type: str = "application/octet-stream",
    tags: dict[str, str] | None = None,
) -> None

Upload data from a bytes buffer or binary stream to a bucket.

Unlike put_object which requires a local file path, this method accepts in-memory bytes or any binary stream, avoiding the need for a temporary file.

Parameters:

Name	Type	Description	Default
bucket_name	str	Destination bucket name.	required
object_name	str	Object name in the bucket.	required
data	bytes | BinaryIO	Content to upload as raw bytes or a binary stream (BinaryIO).	required
length	int	Content length in bytes. If -1, computed automatically for bytes; for streams, providing the exact length avoids buffering overhead.	-1
content_type	str	MIME type of the content. Defaults to "application/octet-stream".	'application/octet-stream'
tags	dict[str, str] | None	Optional key-value tags to attach to the object. Tags can be used with bucket lifecycle rules to implement per-object TTL expiration.	None

Raises:

Type	Description
InvalidArgumentError	If bucket_name, object_name, or data is invalid.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to upload is denied.
ResourceExhaustedError	If storage limits are exceeded.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def put_object_stream(
    self,
    bucket_name: str,
    object_name: str,
    data: bytes | BinaryIO,
    length: int = -1,
    content_type: str = "application/octet-stream",
    tags: dict[str, str] | None = None,
) -> None:
    """Upload data from a bytes buffer or binary stream to a bucket.

    Unlike put_object which requires a local file path, this method accepts
    in-memory bytes or any binary stream, avoiding the need for a temporary file.

    Args:
        bucket_name: Destination bucket name.
        object_name: Object name in the bucket.
        data: Content to upload as raw bytes or a binary stream (BinaryIO).
        length: Content length in bytes. If -1, computed automatically for bytes;
            for streams, providing the exact length avoids buffering overhead.
        content_type: MIME type of the content. Defaults to "application/octet-stream".
        tags: Optional key-value tags to attach to the object. Tags can be used
            with bucket lifecycle rules to implement per-object TTL expiration.

    Raises:
        InvalidArgumentError: If bucket_name, object_name, or data is invalid.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to upload is denied.
        ResourceExhaustedError: If storage limits are exceeded.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )

        kwargs: dict[str, Any] = {
            "Bucket": bucket_name,
            "Key": object_name,
            "Body": data,
            "ContentType": content_type,
        }

        if isinstance(data, bytes):
            kwargs["ContentLength"] = len(data) if length < 0 else length
        elif length >= 0:
            kwargs["ContentLength"] = length

        if tags:
            kwargs["Tagging"] = urllib.parse.urlencode(tags)

        self._client.put_object(**kwargs)
        self._clear_object_caches()
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "put_object_stream")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "put_object_stream")
    except Exception as e:
        self._handle_general_exception(e, "put_object_stream")

archipy.adapters.minio.adapters.MinioAdapter.get_object_stream

get_object_stream(
    bucket_name: str, object_name: str
) -> bytes

Download an object and return its content as bytes.

Unlike get_object which requires a local file path, this method returns the object content directly in memory, avoiding a temporary file.

Parameters:

Name	Type	Description	Default
bucket_name	str	Source bucket name.	required
object_name	str	Object name in the bucket.	required

Returns:

Name	Type	Description
bytes	bytes	The full content of the object.

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to download is denied.
ServiceUnavailableError	If the S3 service is unavailable.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def get_object_stream(self, bucket_name: str, object_name: str) -> bytes:
    """Download an object and return its content as bytes.

    Unlike get_object which requires a local file path, this method returns
    the object content directly in memory, avoiding a temporary file.

    Args:
        bucket_name: Source bucket name.
        object_name: Object name in the bucket.

    Returns:
        bytes: The full content of the object.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to download is denied.
        ServiceUnavailableError: If the S3 service is unavailable.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        response = self._client.get_object(Bucket=bucket_name, Key=object_name)
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "get_object_stream")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "get_object_stream")
        raise
    except Exception as e:
        self._handle_general_exception(e, "get_object_stream")
        raise
    else:
        content: bytes = response["Body"].read()
        return content

archipy.adapters.minio.adapters.MinioAdapter.set_object_tags

set_object_tags(
    bucket_name: str, object_name: str, tags: dict[str, str]
) -> None

Set tags on an existing object, replacing any existing tags.

Tags are key-value pairs that can be used with bucket lifecycle rules to implement TTL-based expiration. For example, tag an object with {"ttl-days": "7"} and create a matching lifecycle rule to auto-delete it after 7 days.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket containing the object.	required
object_name	str	Object name to tag.	required
tags	dict[str, str]	Key-value mapping of tag names to values.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to set tags is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def set_object_tags(self, bucket_name: str, object_name: str, tags: dict[str, str]) -> None:
    """Set tags on an existing object, replacing any existing tags.

    Tags are key-value pairs that can be used with bucket lifecycle rules to
    implement TTL-based expiration. For example, tag an object with
    ``{"ttl-days": "7"}`` and create a matching lifecycle rule to auto-delete
    it after 7 days.

    Args:
        bucket_name: Bucket containing the object.
        object_name: Object name to tag.
        tags: Key-value mapping of tag names to values.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to set tags is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        tag_set = [{"Key": k, "Value": v} for k, v in tags.items()]
        self._client.put_object_tagging(
            Bucket=bucket_name,
            Key=object_name,
            Tagging={"TagSet": tag_set},
        )
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "set_object_tags")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "set_object_tags")
    except Exception as e:
        self._handle_general_exception(e, "set_object_tags")

archipy.adapters.minio.adapters.MinioAdapter.remove_object_tags

remove_object_tags(
    bucket_name: str, object_name: str
) -> None

Remove all tags from an object.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket containing the object.	required
object_name	str	Object name whose tags to remove.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to remove tags is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def remove_object_tags(self, bucket_name: str, object_name: str) -> None:
    """Remove all tags from an object.

    Args:
        bucket_name: Bucket containing the object.
        object_name: Object name whose tags to remove.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to remove tags is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        self._client.delete_object_tagging(Bucket=bucket_name, Key=object_name)
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "remove_object_tags")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "remove_object_tags")
    except Exception as e:
        self._handle_general_exception(e, "remove_object_tags")

archipy.adapters.minio.adapters.MinioAdapter.get_object_tags

get_object_tags(
    bucket_name: str, object_name: str
) -> dict[str, str]

Return the tags attached to an object.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket containing the object.	required
object_name	str	Object name whose tags to retrieve.	required

Returns:

Type	Description
dict[str, str]	dict[str, str]: Mapping of tag key to tag value. Empty dict if no tags.

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket or object does not exist.
PermissionDeniedError	If permission to get tags is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def get_object_tags(self, bucket_name: str, object_name: str) -> dict[str, str]:
    """Return the tags attached to an object.

    Args:
        bucket_name: Bucket containing the object.
        object_name: Object name whose tags to retrieve.

    Returns:
        dict[str, str]: Mapping of tag key to tag value. Empty dict if no tags.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket or object does not exist.
        PermissionDeniedError: If permission to get tags is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name or object_name"
                    if not all([bucket_name, object_name])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                ),
            )
        response = self._client.get_object_tagging(Bucket=bucket_name, Key=object_name)
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "get_object_tags")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "get_object_tags")
        raise
    except Exception as e:
        self._handle_general_exception(e, "get_object_tags")
        raise
    else:
        return {tag["Key"]: tag["Value"] for tag in response.get("TagSet", [])}

archipy.adapters.minio.adapters.MinioAdapter.set_bucket_lifecycle

set_bucket_lifecycle(
    bucket_name: str, rules: list[MinioLifecycleRuleType]
) -> None

Set the lifecycle configuration for a bucket.

Lifecycle rules control automatic expiration (TTL) of objects. Each rule is a dict that maps to the S3 LifecycleRule structure. A minimal expiration rule::

{
    "ID": "expire-after-7-days",
    "Status": "Enabled",
    "Filter": {"Prefix": "uploads/"},
    "Expiration": {"Days": 7},
}

To expire only objects with a specific tag::

{
    "ID": "expire-tagged-7days",
    "Status": "Enabled",
    "Filter": {"Tag": {"Key": "ttl-days", "Value": "7"}},
    "Expiration": {"Days": 7},
}

Parameters:

Name	Type	Description	Default
bucket_name	str	Target bucket name.	required
rules	list[MinioLifecycleRuleType]	List of lifecycle rule dicts (S3 LifecycleRule format).	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty or rules is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to set lifecycle is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def set_bucket_lifecycle(self, bucket_name: str, rules: list[MinioLifecycleRuleType]) -> None:
    """Set the lifecycle configuration for a bucket.

    Lifecycle rules control automatic expiration (TTL) of objects. Each rule is a
    dict that maps to the S3 ``LifecycleRule`` structure. A minimal expiration rule::

        {
            "ID": "expire-after-7-days",
            "Status": "Enabled",
            "Filter": {"Prefix": "uploads/"},
            "Expiration": {"Days": 7},
        }

    To expire only objects with a specific tag::

        {
            "ID": "expire-tagged-7days",
            "Status": "Enabled",
            "Filter": {"Tag": {"Key": "ttl-days", "Value": "7"}},
            "Expiration": {"Days": 7},
        }

    Args:
        bucket_name: Target bucket name.
        rules: List of lifecycle rule dicts (S3 ``LifecycleRule`` format).

    Raises:
        InvalidArgumentError: If bucket_name is empty or rules is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to set lifecycle is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        if not rules:
            raise InvalidArgumentError(argument_name="rules")
        self._client.put_bucket_lifecycle_configuration(
            Bucket=bucket_name,
            LifecycleConfiguration={"Rules": rules},
        )
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "set_bucket_lifecycle")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "set_bucket_lifecycle")
    except Exception as e:
        self._handle_general_exception(e, "set_bucket_lifecycle")

archipy.adapters.minio.adapters.MinioAdapter.get_bucket_lifecycle

get_bucket_lifecycle(
    bucket_name: str,
) -> list[MinioLifecycleRuleType]

Return the lifecycle rules configured for a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required

Returns:

Type	Description
list[MinioLifecycleRuleType]	list[MinioLifecycleRuleType]: List of lifecycle rule dicts, or an empty list if no lifecycle configuration is set.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to get lifecycle is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def get_bucket_lifecycle(self, bucket_name: str) -> list[MinioLifecycleRuleType]:
    """Return the lifecycle rules configured for a bucket.

    Args:
        bucket_name: Bucket name.

    Returns:
        list[MinioLifecycleRuleType]: List of lifecycle rule dicts, or an empty
            list if no lifecycle configuration is set.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to get lifecycle is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        response = self._client.get_bucket_lifecycle_configuration(Bucket=bucket_name)
    except InvalidArgumentError:
        raise
    except ClientError as e:
        if e.response.get("Error", {}).get("Code", "") == "NoSuchLifecycleConfiguration":
            return []
        self._handle_client_exception(e, "get_bucket_lifecycle")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "get_bucket_lifecycle")
        raise
    except Exception as e:
        self._handle_general_exception(e, "get_bucket_lifecycle")
        raise
    else:
        rules: list[MinioLifecycleRuleType] = response.get("Rules", [])
        return rules

archipy.adapters.minio.adapters.MinioAdapter.delete_bucket_lifecycle

delete_bucket_lifecycle(bucket_name: str) -> None

Delete the lifecycle configuration from a bucket.

After this call the bucket has no lifecycle rules and objects will not be automatically expired.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to delete lifecycle is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def delete_bucket_lifecycle(self, bucket_name: str) -> None:
    """Delete the lifecycle configuration from a bucket.

    After this call the bucket has no lifecycle rules and objects will not be
    automatically expired.

    Args:
        bucket_name: Bucket name.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to delete lifecycle is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        self._client.delete_bucket_lifecycle(Bucket=bucket_name)
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "delete_bucket_lifecycle")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "delete_bucket_lifecycle")
    except Exception as e:
        self._handle_general_exception(e, "delete_bucket_lifecycle")

archipy.adapters.minio.adapters.MinioAdapter.set_bucket_versioning

set_bucket_versioning(
    bucket_name: str, *, enabled: bool
) -> None

Enable or suspend versioning on a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
enabled	bool	If True, enable versioning; if False, suspend it.	required

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to set versioning is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def set_bucket_versioning(self, bucket_name: str, *, enabled: bool) -> None:
    """Enable or suspend versioning on a bucket.

    Args:
        bucket_name: Bucket name.
        enabled: If True, enable versioning; if False, suspend it.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to set versioning is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        status = "Enabled" if enabled else "Suspended"
        self._client.put_bucket_versioning(
            Bucket=bucket_name,
            VersioningConfiguration={"Status": status},
        )
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "set_bucket_versioning")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "set_bucket_versioning")
    except Exception as e:
        self._handle_general_exception(e, "set_bucket_versioning")

archipy.adapters.minio.adapters.MinioAdapter.get_bucket_versioning

get_bucket_versioning(bucket_name: str) -> str

Return the versioning status of a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required

Returns:

Name	Type	Description
str	str	Versioning status — Enabled, Suspended, or empty string if never configured.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to get versioning is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)
def get_bucket_versioning(self, bucket_name: str) -> str:
    """Return the versioning status of a bucket.

    Args:
        bucket_name: Bucket name.

    Returns:
        str: Versioning status — ``Enabled``, ``Suspended``, or empty string if never configured.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to get versioning is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")
        response = self._client.get_bucket_versioning(Bucket=bucket_name)
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "get_bucket_versioning")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "get_bucket_versioning")
        raise
    except Exception as e:
        self._handle_general_exception(e, "get_bucket_versioning")
        raise
    else:
        status: str = response.get("Status", "")
        return status

archipy.adapters.minio.adapters.MinioAdapter.list_object_versions

list_object_versions(
    bucket_name: str, prefix: str = ""
) -> list[MinioObjectVersionType]

List all versions and delete markers for objects in a bucket.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
prefix	str	Optional prefix to filter objects by.	''

Returns:

Type	Description
list[MinioObjectVersionType]	list[MinioObjectVersionType]: Version entries with keys, version IDs, and metadata.

Raises:

Type	Description
InvalidArgumentError	If bucket_name is empty.
NotFoundError	If the bucket does not exist.
PermissionDeniedError	If permission to list versions is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
@ttl_cache_decorator(ttl_seconds=300, maxsize=100)
def list_object_versions(
    self,
    bucket_name: str,
    prefix: str = "",
) -> list[MinioObjectVersionType]:
    """List all versions and delete markers for objects in a bucket.

    Args:
        bucket_name: Bucket name.
        prefix: Optional prefix to filter objects by.

    Returns:
        list[MinioObjectVersionType]: Version entries with keys, version IDs, and metadata.

    Raises:
        InvalidArgumentError: If bucket_name is empty.
        NotFoundError: If the bucket does not exist.
        PermissionDeniedError: If permission to list versions is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name:
            raise InvalidArgumentError(argument_name="bucket_name")

        params: dict[str, str] = {"Bucket": bucket_name}
        if prefix:
            params["Prefix"] = prefix

        version_list: list[MinioObjectVersionType] = []
        paginator = self._client.get_paginator("list_object_versions")
        for page in paginator.paginate(**params):
            version_list.extend(
                {
                    "object_name": version["Key"],
                    "version_id": version["VersionId"],
                    "is_latest": version.get("IsLatest", False),
                    "is_delete_marker": False,
                    "size": version.get("Size", 0),
                    "last_modified": version.get("LastModified"),
                }
                for version in page.get("Versions", [])
            )
            version_list.extend(
                {
                    "object_name": marker["Key"],
                    "version_id": marker["VersionId"],
                    "is_latest": marker.get("IsLatest", False),
                    "is_delete_marker": True,
                    "size": 0,
                    "last_modified": marker.get("LastModified"),
                }
                for marker in page.get("DeleteMarkers", [])
            )
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "list_object_versions")
        raise
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "list_object_versions")
        raise
    except Exception as e:
        self._handle_general_exception(e, "list_object_versions")
        raise
    else:
        return version_list

archipy.adapters.minio.adapters.MinioAdapter.remove_object_version

remove_object_version(
    bucket_name: str, object_name: str, version_id: str
) -> None

Permanently delete a specific object version.

Parameters:

Name	Type	Description	Default
bucket_name	str	Bucket name.	required
object_name	str	Object key.	required
version_id	str	Version ID to delete permanently.	required

Raises:

Type	Description
InvalidArgumentError	If any required parameter is empty.
NotFoundError	If the bucket, object, or version does not exist.
PermissionDeniedError	If permission to delete the version is denied.
StorageError	If there's a storage-related error.

Source code in archipy/adapters/minio/adapters.py

@override
def remove_object_version(self, bucket_name: str, object_name: str, version_id: str) -> None:
    """Permanently delete a specific object version.

    Args:
        bucket_name: Bucket name.
        object_name: Object key.
        version_id: Version ID to delete permanently.

    Raises:
        InvalidArgumentError: If any required parameter is empty.
        NotFoundError: If the bucket, object, or version does not exist.
        PermissionDeniedError: If permission to delete the version is denied.
        StorageError: If there's a storage-related error.
    """
    try:
        if not bucket_name or not object_name or not version_id:
            raise InvalidArgumentError(
                argument_name=(
                    "bucket_name, object_name or version_id"
                    if not all([bucket_name, object_name, version_id])
                    else "bucket_name"
                    if not bucket_name
                    else "object_name"
                    if not object_name
                    else "version_id"
                ),
            )
        self._client.delete_object(Bucket=bucket_name, Key=object_name, VersionId=version_id)
        self._clear_object_caches()
    except InvalidArgumentError:
        raise
    except ClientError as e:
        self._handle_client_exception(e, "remove_object_version")
    except (ConnectionError, EndpointConnectionError) as e:
        self._handle_connection_exception(e, "remove_object_version")
    except Exception as e:
        self._handle_general_exception(e, "remove_object_version")

options: show_root_toc_entry: false heading_level: 3