tessl/pypi-minio

MinIO Python SDK for Amazon S3 Compatible Cloud Storage

—

Pending

Overview

Eval results

Files

Configuration Classes

Name: tessl/pypi-minio
Author: tessl

Configuration objects for bucket settings, server-side encryption, lifecycle management, object policies, versioning, and other advanced S3 features. These classes provide type-safe configuration for sophisticated object storage operations.

Capabilities

Server-Side Encryption (SSE)

Configure encryption for objects and buckets with support for customer keys, AWS KMS, and S3-managed encryption.

class Sse:
    """Base class for server-side encryption configurations."""
    def headers(self) -> dict[str, str]:
        """
        Return HTTP headers for encryption configuration.
        
        Returns:
            Dictionary of headers for S3 requests
        """
    
    def tls_required(self) -> bool:
        """
        Check if TLS is required for this encryption method.
        
        Returns:
            True if TLS/HTTPS is required
        """
    
    def copy_headers(self) -> dict[str, str]:
        """
        Return HTTP headers for copy operations.
        
        Returns:
            Dictionary of headers for S3 copy requests
        """

class SseCustomerKey(Sse):
    """Server-side encryption with customer-provided keys (SSE-C)."""
    def __init__(self, key: bytes) -> None:
        """
        Initialize SSE-C encryption.
        
        Args:
            key: 32-byte AES256 encryption key
            
        Raises:
            ValueError: If key is not exactly 32 bytes
        """
    
    def headers(self) -> dict[str, str]:
        """
        Return SSE-C headers.
        
        Returns:
            Headers with customer encryption key and algorithm
        """

class SseKMS(Sse):
    """Server-side encryption with AWS Key Management Service."""
    def __init__(
        self, 
        key: str, 
        context: dict[str, str] | None = None
    ) -> None:
        """
        Initialize SSE-KMS encryption.
        
        Args:
            key: KMS key ID or ARN
            context: Encryption context key-value pairs (optional)
        """
    
    def headers(self) -> dict[str, str]:
        """
        Return SSE-KMS headers.
        
        Returns:
            Headers with KMS key and encryption context
        """

class SseS3(Sse):
    """Server-side encryption with S3-managed keys (SSE-S3)."""
    def __init__(self) -> None:
        """Initialize SSE-S3 encryption with S3-managed keys."""
    
    def headers(self) -> dict[str, str]:
        """
        Return SSE-S3 headers.
        
        Returns:
            Headers for S3-managed encryption
        """

class SSEConfig:
    """Bucket default encryption configuration."""
    def __init__(self, rule: SSERule) -> None:
        """
        Initialize bucket encryption configuration.
        
        Args:
            rule: SSE rule defining encryption settings
        """
    rule: SSERule
    
    @classmethod
    def fromxml(cls, element: Element) -> SSEConfig:
        """Create SSEConfig from XML element."""
    
    def toxml(self, element: Element) -> Element:
        """Convert SSEConfig to XML element."""

class SSERule:
    """Rule for bucket default encryption."""
    def __init__(
        self,
        sse_algorithm: str,
        kms_master_key_id: str | None = None
    ) -> None:
        """
        Initialize SSE rule.
        
        Args:
            sse_algorithm: Encryption algorithm ("AES256" or "aws:kms")
            kms_master_key_id: KMS key ID (required for aws:kms)
        """
    sse_algorithm: str
    kms_master_key_id: str | None

Tags and Metadata

Manage object and bucket tags for organization, billing, and lifecycle management.

class Tags(dict[str, str]):
    """Dictionary extended for bucket/object tags with validation."""
    def __init__(self, for_object: bool = False) -> None:
        """
        Initialize tags container.
        
        Args:
            for_object: True for object tags (max 10), False for bucket tags (max 50)
        """
    
    @classmethod
    def new_bucket_tags(cls) -> Tags:
        """
        Create new bucket tags instance.
        
        Returns:
            Tags instance configured for bucket use (max 50 tags)
        """
    
    @classmethod
    def new_object_tags(cls) -> Tags:
        """
        Create new object tags instance.
        
        Returns:
            Tags instance configured for object use (max 10 tags)
        """
    
    @classmethod
    def fromxml(cls, element: Element) -> Tags:
        """Create Tags from XML element."""
    
    def toxml(self, element: Element) -> Element:
        """Convert Tags to XML element."""

Object Copy and Compose Operations

Configuration classes for advanced object operations including copying and composing multiple sources.

class CopySource:
    """Copy source specification for copy operations."""
    def __init__(
        self,
        bucket_name: str,
        object_name: str,
        version_id: str | None = None,
        ssec: SseCustomerKey | None = None,
        offset: int | None = None,
        length: int | None = None,
        match_etag: str | None = None,
        not_match_etag: str | None = None,
        modified_since: datetime.datetime | None = None,
        unmodified_since: datetime.datetime | None = None
    ) -> None:
        """
        Initialize copy source.
        
        Args:
            bucket_name: Source bucket name
            object_name: Source object name
            version_id: Specific version to copy (optional)
            ssec: Server-side encryption key for encrypted source
            offset: Byte offset to start copying from
            length: Number of bytes to copy
            match_etag: Copy only if ETag matches
            not_match_etag: Copy only if ETag doesn't match
            modified_since: Copy only if modified after this date
            unmodified_since: Copy only if not modified after this date
        """
    bucket_name: str
    object_name: str
    version_id: str | None
    ssec: SseCustomerKey | None
    offset: int | None
    length: int | None

class ComposeSource:
    """Compose source specification for compose operations."""
    def __init__(
        self,
        bucket_name: str,
        object_name: str,
        version_id: str | None = None,
        ssec: SseCustomerKey | None = None,
        offset: int | None = None,
        length: int | None = None,
        match_etag: str | None = None,
        not_match_etag: str | None = None,
        modified_since: datetime.datetime | None = None,
        unmodified_since: datetime.datetime | None = None
    ) -> None:
        """
        Initialize compose source.
        
        Args:
            bucket_name: Source bucket name
            object_name: Source object name
            version_id: Specific version to include (optional)
            ssec: Server-side encryption key for encrypted source
            offset: Byte offset to start including from
            length: Number of bytes to include
            match_etag: Include only if ETag matches
            not_match_etag: Include only if ETag doesn't match
            modified_since: Include only if modified after this date
            unmodified_since: Include only if not modified after this date
        """
    bucket_name: str
    object_name: str
    version_id: str | None
    ssec: SseCustomerKey | None
    offset: int | None
    length: int | None

Versioning Configuration

Configure bucket versioning to maintain multiple versions of objects.

class VersioningConfig:
    """Bucket versioning configuration."""
    
    # Constants for versioning status
    ENABLED: str = "Enabled"
    SUSPENDED: str = "Suspended"
    OFF: str = "Off"
    
    def __init__(
        self, 
        status: str | None = None, 
        mfa_delete: str | None = None
    ) -> None:
        """
        Initialize versioning configuration.
        
        Args:
            status: Versioning status (ENABLED, SUSPENDED, or OFF)
            mfa_delete: MFA delete requirement (ENABLED or DISABLED)
        """
    status: str | None
    mfa_delete: str | None
    
    @classmethod
    def fromxml(cls, element: Element) -> VersioningConfig:
        """Create VersioningConfig from XML element."""
    
    def toxml(self, element: Element) -> Element:
        """Convert VersioningConfig to XML element."""

Lifecycle Management

Configure automated lifecycle policies for object transitions and expiration.

class LifecycleConfig:
    """Bucket lifecycle configuration with rules."""
    def __init__(self, rules: list[Rule]) -> None:
        """
        Initialize lifecycle configuration.
        
        Args:
            rules: List of lifecycle rules
        """
    rules: list[Rule]
    
    @classmethod
    def fromxml(cls, element: Element) -> LifecycleConfig:
        """Create LifecycleConfig from XML element."""
    
    def toxml(self, element: Element) -> Element:
        """Convert LifecycleConfig to XML element."""

class Rule:
    """Individual lifecycle rule."""
    def __init__(
        self,
        rule_id: str,
        status: str,
        rule_filter: Filter | None = None,
        expiration: Expiration | None = None,
        noncurrent_version_expiration: NoncurrentVersionExpiration | None = None,
        abort_incomplete_multipart_upload: AbortIncompleteMultipartUpload | None = None,
        transition: Transition | None = None,
        noncurrent_version_transition: NoncurrentVersionTransition | None = None
    ) -> None:
        """
        Initialize lifecycle rule.
        
        Args:
            rule_id: Unique identifier for the rule
            status: Rule status ("Enabled" or "Disabled")
            rule_filter: Filter to determine which objects the rule applies to
            expiration: Object expiration configuration
            noncurrent_version_expiration: Non-current version expiration
            abort_incomplete_multipart_upload: Multipart upload cleanup
            transition: Storage class transition configuration
            noncurrent_version_transition: Non-current version transitions
        """
    rule_id: str
    status: str
    rule_filter: Filter | None
    expiration: Expiration | None

class Filter:
    """Filter for lifecycle rules."""
    def __init__(
        self,
        prefix: str | None = None,
        tag: Tag | None = None,
        and_filter: And | None = None
    ) -> None:
        """Initialize lifecycle filter."""
    prefix: str | None
    tag: Tag | None
    and_filter: And | None

class Expiration:
    """Object expiration configuration."""
    def __init__(
        self,
        date: datetime.date | None = None,
        days: int | None = None,
        expired_object_delete_marker: bool | None = None
    ) -> None:
        """
        Initialize expiration configuration.
        
        Args:
            date: Specific expiration date
            days: Days after object creation to expire
            expired_object_delete_marker: Remove expired delete markers
        """
    date: datetime.date | None
    days: int | None
    expired_object_delete_marker: bool | None

class Transition:
    """Storage class transition configuration."""
    def __init__(
        self,
        date: datetime.date | None = None,
        days: int | None = None,
        storage_class: str | None = None
    ) -> None:
        """
        Initialize transition configuration.
        
        Args:
            date: Specific transition date
            days: Days after object creation to transition
            storage_class: Target storage class (GLACIER, DEEP_ARCHIVE, etc.)
        """
    date: datetime.date | None
    days: int | None
    storage_class: str | None

Bucket Notifications

Configure bucket notifications for real-time event handling.

class NotificationConfig:
    """Bucket notification configuration."""
    def __init__(
        self,
        cloud_func_config_list: list[CloudFunctionConfig] | None = None,
        queue_config_list: list[QueueConfig] | None = None,
        topic_config_list: list[TopicConfig] | None = None
    ) -> None:
        """
        Initialize notification configuration.
        
        Args:
            cloud_func_config_list: Lambda function configurations
            queue_config_list: SQS queue configurations  
            topic_config_list: SNS topic configurations
        """
    cloud_func_config_list: list[CloudFunctionConfig] | None
    queue_config_list: list[QueueConfig] | None
    topic_config_list: list[TopicConfig] | None

class CloudFunctionConfig:
    """Lambda function notification configuration."""
    def __init__(
        self,
        id_: str,
        cloud_func: str,
        events: list[str],
        prefix_filter_rule: PrefixFilterRule | None = None,
        suffix_filter_rule: SuffixFilterRule | None = None
    ) -> None:
        """
        Initialize Lambda function configuration.
        
        Args:
            id_: Unique configuration identifier
            cloud_func: Lambda function ARN
            events: List of S3 events to trigger on
            prefix_filter_rule: Object key prefix filter
            suffix_filter_rule: Object key suffix filter
        """

class QueueConfig:
    """SQS queue notification configuration."""
    def __init__(
        self,
        id_: str,
        queue: str,
        events: list[str],
        prefix_filter_rule: PrefixFilterRule | None = None,
        suffix_filter_rule: SuffixFilterRule | None = None
    ) -> None:
        """Initialize SQS queue configuration."""

class TopicConfig:
    """SNS topic notification configuration."""
    def __init__(
        self,
        id_: str,
        topic: str,
        events: list[str],
        prefix_filter_rule: PrefixFilterRule | None = None,
        suffix_filter_rule: SuffixFilterRule | None = None
    ) -> None:
        """Initialize SNS topic configuration."""

Object Lock and Retention

Configure object lock for compliance and legal hold requirements.

class ObjectLockConfig:
    """Object lock configuration for compliance."""
    def __init__(
        self,
        retention_mode: str | None = None,
        retention_duration_days: int | None = None,
        retention_duration_years: int | None = None
    ) -> None:
        """
        Initialize object lock configuration.
        
        Args:
            retention_mode: Retention mode ("GOVERNANCE" or "COMPLIANCE")
            retention_duration_days: Retention period in days
            retention_duration_years: Retention period in years
        """
    retention_mode: str | None
    retention_duration_days: int | None
    retention_duration_years: int | None

class Retention:
    """Object retention configuration."""
    def __init__(
        self, 
        mode: str, 
        retain_until_date: datetime.datetime
    ) -> None:
        """
        Initialize retention configuration.
        
        Args:
            mode: Retention mode ("GOVERNANCE" or "COMPLIANCE")
            retain_until_date: Date until which object is retained
        """
    mode: str
    retain_until_date: datetime.datetime
    
    @classmethod
    def fromxml(cls, element: Element) -> Retention:
        """Create Retention from XML element."""
    
    def toxml(self, element: Element) -> Element:
        """Convert Retention to XML element."""

class LegalHold:
    """Legal hold configuration for objects."""
    def __init__(self, status: bool) -> None:
        """
        Initialize legal hold configuration.
        
        Args:
            status: True to enable legal hold, False to disable
        """
    status: bool
    
    @classmethod
    def fromxml(cls, element: Element) -> LegalHold:
        """Create LegalHold from XML element."""
    
    def toxml(self, element: Element) -> Element:
        """Convert LegalHold to XML element."""

Replication Configuration

Configure cross-region replication for disaster recovery and compliance.

class ReplicationConfig:
    """Bucket replication configuration."""
    def __init__(self, role: str, rules: list[ReplicationRule]) -> None:
        """
        Initialize replication configuration.
        
        Args:
            role: IAM role ARN for replication
            rules: List of replication rules
        """
    role: str
    rules: list[ReplicationRule]

class ReplicationRule:
    """Individual replication rule."""
    def __init__(
        self,
        rule_id: str,
        status: str,
        priority: int,
        delete_marker_replication: DeleteMarkerReplication | None = None,
        rule_filter: Filter | None = None,
        destination: Destination | None = None
    ) -> None:
        """
        Initialize replication rule.
        
        Args:
            rule_id: Unique identifier for the rule
            status: Rule status ("Enabled" or "Disabled")
            priority: Rule priority (higher numbers have precedence)
            delete_marker_replication: Delete marker replication settings
            rule_filter: Filter to determine which objects to replicate
            destination: Replication destination configuration
        """
    rule_id: str
    status: str
    priority: int
    destination: Destination | None

class Destination:
    """Replication destination configuration."""
    def __init__(
        self,
        bucket_arn: str,
        storage_class: str | None = None,
        access_control_translation: AccessControlTranslation | None = None,
        encryption_configuration: EncryptionConfiguration | None = None
    ) -> None:
        """
        Initialize replication destination.
        
        Args:
            bucket_arn: Destination bucket ARN
            storage_class: Target storage class
            access_control_translation: Access control settings
            encryption_configuration: Encryption settings for destination
        """
    bucket_arn: str
    storage_class: str | None

Usage Examples

Server-Side Encryption

from minio import Minio
from minio.sse import SseCustomerKey, SseKMS, SseS3
import secrets

client = Minio("localhost:9000", "minio", "minio123")

# Customer-provided key encryption (SSE-C)
customer_key = secrets.token_bytes(32)  # 32-byte AES key
sse_c = SseCustomerKey(customer_key)

client.put_object(
    "my-bucket", 
    "encrypted-file.txt", 
    data=io.BytesIO(b"sensitive data"),
    length=13,
    sse=sse_c
)

# KMS encryption (SSE-KMS)
sse_kms = SseKMS("arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012")

client.put_object(
    "my-bucket",
    "kms-encrypted.txt",
    data=io.BytesIO(b"kms protected data"),
    length=18,
    sse=sse_kms
)

# S3-managed encryption (SSE-S3)
sse_s3 = SseS3()

client.put_object(
    "my-bucket",
    "s3-encrypted.txt", 
    data=io.BytesIO(b"s3 protected data"),
    length=17,
    sse=sse_s3
)

Tags and Metadata

from minio.commonconfig import Tags

# Create bucket tags
bucket_tags = Tags.new_bucket_tags()
bucket_tags["Environment"] = "Production"
bucket_tags["Team"] = "Engineering"
bucket_tags["Cost-Center"] = "Engineering-2024"

client.set_bucket_tags("my-bucket", bucket_tags)

# Create object tags
object_tags = Tags.new_object_tags()
object_tags["Type"] = "Document"
object_tags["Classification"] = "Internal"

client.put_object(
    "my-bucket",
    "document.pdf",
    data=io.BytesIO(b"document content"),
    length=16,
    tags=object_tags
)

# Retrieve and modify tags
existing_tags = client.get_bucket_tags("my-bucket")
if existing_tags:
    existing_tags["LastModified"] = datetime.datetime.now().isoformat()
    client.set_bucket_tags("my-bucket", existing_tags)

Bucket Versioning

from minio.versioningconfig import VersioningConfig

# Enable versioning
versioning_config = VersioningConfig(VersioningConfig.ENABLED)
client.set_bucket_versioning("my-bucket", versioning_config)

# Check versioning status
current_versioning = client.get_bucket_versioning("my-bucket")
print(f"Versioning status: {current_versioning.status}")

# Suspend versioning (keeps existing versions)
suspend_config = VersioningConfig(VersioningConfig.SUSPENDED)
client.set_bucket_versioning("my-bucket", suspend_config)

Lifecycle Management

from minio.lifecycleconfig import (
    LifecycleConfig, Rule, Filter, Expiration, Transition
)
import datetime

# Create lifecycle rules
rules = []

# Rule 1: Delete objects after 365 days
expiration_rule = Rule(
    rule_id="delete-old-objects",
    status="Enabled",
    rule_filter=Filter(prefix="logs/"),
    expiration=Expiration(days=365)
)
rules.append(expiration_rule)

# Rule 2: Transition to Glacier after 30 days
transition_rule = Rule(
    rule_id="archive-documents", 
    status="Enabled",
    rule_filter=Filter(prefix="documents/"),
    transition=Transition(days=30, storage_class="GLACIER")
)
rules.append(transition_rule)

# Apply lifecycle configuration
lifecycle_config = LifecycleConfig(rules)
client.set_bucket_lifecycle("my-bucket", lifecycle_config)

# Get current lifecycle configuration
current_lifecycle = client.get_bucket_lifecycle("my-bucket")
for rule in current_lifecycle.rules:
    print(f"Rule: {rule.rule_id}, Status: {rule.status}")

Object Copy with Advanced Options

from minio.commonconfig import CopySource, Tags
from minio.sse import SseS3
import datetime

# Copy with conditions and encryption
source = CopySource(
    bucket_name="source-bucket",
    object_name="source-object.txt",
    modified_since=datetime.datetime(2024, 1, 1)  # Only copy if modified after date
)

# Copy with new metadata and encryption
new_tags = Tags.new_object_tags()
new_tags["Source"] = "Migrated"
new_tags["Date"] = datetime.datetime.now().isoformat()

result = client.copy_object(
    bucket_name="dest-bucket",
    object_name="dest-object.txt", 
    source=source,
    sse=SseS3(),
    metadata={"Content-Type": "text/plain", "Author": "System"},
    tags=new_tags
)

print(f"Copied object ETag: {result.etag}")

Object Retention and Legal Hold

from minio.retention import Retention
from minio.legalhold import LegalHold
import datetime

# Set retention policy (WORM - Write Once Read Many)
retention_date = datetime.datetime.utcnow() + datetime.timedelta(days=2555)  # 7 years
retention = Retention("COMPLIANCE", retention_date)

# Upload with retention
client.put_object(
    "compliance-bucket",
    "compliance-document.pdf",
    data=io.BytesIO(b"compliance document"),
    length=19,
    retention=retention,
    legal_hold=True
)

# Modify legal hold status
legal_hold = LegalHold(False)  # Remove legal hold
client.set_object_legal_hold(
    "compliance-bucket", 
    "compliance-document.pdf", 
    legal_hold
)

Bucket Default Encryption

from minio.sseconfig import SSEConfig, SSERule

# Set bucket default encryption to SSE-S3
sse_rule = SSERule("AES256")
sse_config = SSEConfig(sse_rule)
client.set_bucket_encryption("my-bucket", sse_config)

# Set bucket default encryption to KMS
kms_rule = SSERule("aws:kms", "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012")
kms_config = SSEConfig(kms_rule)
client.set_bucket_encryption("kms-bucket", kms_config)

# Get current encryption configuration
current_encryption = client.get_bucket_encryption("my-bucket")
if current_encryption:
    print(f"Encryption algorithm: {current_encryption.rule.sse_algorithm}")

Install with Tessl CLI