or run

npx @tessl/cli init

Version

Tile

Overview

Evals

Files

docs

admin-operations.md advanced-operations.md client-operations.md configuration.md credentials-auth.md error-handling.md index.md

tile.json

tessl/pypi-minio

MinIO Python SDK for Amazon S3 Compatible Cloud Storage

Workspace: tessl
Visibility: Public
Created: 3 months ago
Last updated: 3 months ago
Describes: pkg:pypi/minio@7.2.x

To install, run

npx @tessl/cli install tessl/pypi-minio@7.2.0

MinIO Python SDK

MinIO Python SDK provides simple APIs to access any Amazon S3 Compatible Object Storage. It supports full S3 API compatibility with advanced features like server-side encryption, object versioning, lifecycle management, and bucket notifications. The SDK includes both standard S3 operations via the Minio class and administrative operations via the MinioAdmin class.

Package Information

Package Name: minio
Language: Python
Installation: pip install minio
Version: 7.2.16

Core Imports

from minio import Minio, MinioAdmin

For specific functionality:

from minio import (
    Minio, MinioAdmin,
    S3Error, InvalidResponseError, ServerError
)
from minio.credentials import StaticProvider, EnvAWSProvider, ChainedProvider
from minio.sse import SseCustomerKey, SseKMS, SseS3
from minio.commonconfig import Tags, CopySource, ComposeSource

Basic Usage

from minio import Minio
from minio.error import S3Error

# Create MinIO client
client = Minio(
    "play.min.io",
    access_key="minioadmin",
    secret_key="minioadmin",
    secure=True
)

# Create bucket
try:
    client.make_bucket("my-bucket")
except S3Error as e:
    print(f"Error: {e}")

# Upload object
try:
    result = client.fput_object(
        "my-bucket", "my-object", "/path/to/file.txt"
    )
    print(f"Uploaded: {result.object_name}")
except S3Error as e:
    print(f"Error: {e}")

# Download object
try:
    client.fget_object("my-bucket", "my-object", "/path/to/download.txt")
    print("Downloaded successfully")
except S3Error as e:
    print(f"Error: {e}")

# List objects
try:
    objects = client.list_objects("my-bucket")
    for obj in objects:
        print(f"Object: {obj.object_name}, Size: {obj.size}")
except S3Error as e:
    print(f"Error: {e}")

Architecture

The MinIO Python SDK is built around two main client classes:

Minio: Standard S3 operations including bucket management, object operations, versioning, lifecycle, encryption, and policies
MinioAdmin: Administrative operations for MinIO server management including user/group management, policies, service accounts, and server configuration

Thread Safety: Both clients are thread-safe when using Python's threading library but are NOT safe for multiprocessing.

Credential Management: Flexible credential system supporting static credentials, environment variables, AWS config files, IAM roles, and credential chaining.

Error Handling: Comprehensive exception hierarchy with S3Error for S3 operations, ServerError for HTTP errors, and InvalidResponseError for malformed responses.

Capabilities

Client Operations

Core bucket and object operations including CRUD operations, metadata management, and basic configuration.

class Minio:
    def __init__(
        self,
        endpoint: str,
        access_key: str | None = None,
        secret_key: str | None = None,
        session_token: str | None = None,
        secure: bool = True,
        region: str | None = None,
        http_client: urllib3.PoolManager | None = None,
        credentials: Provider | None = None,
        cert_check: bool = True
    ) -> None: ...
    
    def make_bucket(
        self, 
        bucket_name: str, 
        location: str | None = None, 
        object_lock: bool = False
    ) -> None: ...
    
    def list_buckets(self) -> list[Bucket]: ...
    
    def put_object(
        self,
        bucket_name: str,
        object_name: str, 
        data: io.IOBase,
        length: int = -1,
        content_type: str = "application/octet-stream",
        metadata: dict[str, str] | None = None,
        sse: Sse | None = None,
        progress: ProgressType | None = None,
        part_size: int = 0,
        num_parallel_uploads: int = 3,
        tags: Tags | None = None,
        retention: Retention | None = None,
        legal_hold: bool = False
    ) -> ObjectWriteResult: ...
    
    def get_object(
        self,
        bucket_name: str,
        object_name: str,
        offset: int = 0,
        length: int = 0,
        request_headers: dict[str, str] | None = None,
        ssec: SseCustomerKey | None = None,
        version_id: str | None = None,
        extra_query_params: dict[str, str] | None = None
    ) -> urllib3.HTTPResponse: ...
    
    def prompt_object(
        self,
        bucket_name: str,
        object_name: str,
        prompt: str,
        lambda_arn: str | None = None,
        request_headers: dict[str, str] | None = None,
        ssec: SseCustomerKey | None = None,
        version_id: str | None = None,
        **kwargs: Any
    ) -> urllib3.HTTPResponse: ...
    
    def enable_object_legal_hold(
        self,
        bucket_name: str,
        object_name: str,
        version_id: str | None = None
    ) -> None: ...
    
    def disable_object_legal_hold(
        self,
        bucket_name: str,
        object_name: str,
        version_id: str | None = None
    ) -> None: ...
    
    def is_object_legal_hold_enabled(
        self,
        bucket_name: str,
        object_name: str,
        version_id: str | None = None
    ) -> bool: ...

Client Operations

Advanced Operations

Advanced S3 features including presigned URLs, SQL SELECT queries on objects, and bucket notifications.

def presigned_get_object(
    self,
    bucket_name: str,
    object_name: str,
    expires: datetime.timedelta = datetime.timedelta(days=7),
    response_headers: dict[str, str] | None = None,
    request_date: datetime.datetime | None = None,
    version_id: str | None = None,
    extra_query_params: dict[str, str] | None = None
) -> str: ...

def select_object_content(
    self,
    bucket_name: str,
    object_name: str,
    request: SelectRequest
) -> SelectObjectReader: ...

def listen_bucket_notification(
    self,
    bucket_name: str,
    prefix: str = "",
    suffix: str = "",
    events: list[str] = ["s3:ObjectCreated:*", "s3:ObjectRemoved:*"]
) -> EventIterable: ...

Advanced Operations

Administrative Operations

MinIO server administration including user management, policy configuration, and service accounts.

class MinioAdmin:
    def __init__(
        self,
        endpoint: str,
        credentials: Provider,
        region: str = "",
        secure: bool = True,
        cert_check: bool = True,
        http_client: urllib3.PoolManager | None = None
    ) -> None: ...
    
    def user_add(self, access_key: str, secret_key: str) -> str: ...
    def policy_add(self, policy_name: str, policy_file: str) -> str: ...
    def info(self) -> str: ...
    def service_restart(self) -> str: ...
    def profile_start(self, profilers: tuple[str] = ()) -> str: ...
    def get_data_usage_info(self) -> str: ...
    def bucket_quota_clear(self, bucket: str) -> str: ...
    def add_site_replication(self, peer_sites: list[PeerSite]) -> str: ...
    def get_site_replication_info(self) -> str: ...
    def attach_policy_ldap(
        self,
        policies: list[str],
        user: str | None = None,
        group: str | None = None
    ) -> str: ...
    def attach_policy(
        self,
        policies: list[str],
        user: str | None = None,
        group: str | None = None
    ) -> str: ...

Administrative Operations

Credentials and Authentication

Comprehensive credential management system with multiple provider types and chaining support.

class Credentials:
    def __init__(
        self,
        access_key: str,
        secret_key: str, 
        session_token: str | None = None,
        expiration: datetime.datetime | None = None
    ) -> None: ...

class StaticProvider(Provider):
    def __init__(
        self,
        access_key: str,
        secret_key: str,
        session_token: str | None = None
    ) -> None: ...

class ChainedProvider(Provider):
    def __init__(self, providers: list[Provider]) -> None: ...

Credentials and Authentication

Configuration Classes

Configuration objects for bucket settings, server-side encryption, lifecycle management, and object policies.

class Tags(dict):
    @classmethod
    def new_bucket_tags(cls) -> Tags: ...
    @classmethod  
    def new_object_tags(cls) -> Tags: ...

class VersioningConfig:
    def __init__(
        self,
        status: str | None = None,
        mfa_delete: str | None = None
    ) -> None: ...

class SseCustomerKey(Sse):
    def __init__(self, key: bytes) -> None: ...

class SseKMS(Sse):
    def __init__(self, key: str, context: dict[str, str] | None = None) -> None: ...

Configuration Classes

Error Handling

Exception hierarchy for comprehensive error handling across all MinIO operations.

class MinioException(Exception):
    """Base exception class for MinIO operations."""

class S3Error(MinioException):
    def __init__(
        self,
        code: str,
        message: str,
        resource: str | None = None,
        request_id: str | None = None,
        host_id: str | None = None,
        response: urllib3.HTTPResponse | None = None,
        bucket_name: str | None = None,
        object_name: str | None = None
    ) -> None: ...

class InvalidResponseError(MinioException):
    def __init__(
        self, 
        code: int | None = None, 
        content_type: str | None = None, 
        body: str | None = None
    ) -> None: ...

class ServerError(MinioException):
    def __init__(self, message: str, status_code: int) -> None: ...

Error Handling

Types

Core Data Types

class Bucket:
    def __init__(self, name: str, creation_date: datetime.datetime | None = None) -> None: ...
    name: str
    creation_date: datetime.datetime | None

class Object:
    bucket_name: str | None
    object_name: str | None
    last_modified: datetime.datetime | None
    etag: str | None
    size: int | None
    content_type: str | None
    is_dir: bool
    version_id: str | None
    is_latest: bool
    is_delete_marker: bool
    storage_class: str | None
    owner_id: str | None
    owner_name: str | None
    tags: Tags | None

class ObjectWriteResult:
    def __init__(
        self,
        bucket_name: str,
        object_name: str,
        etag: str,
        version_id: str | None = None,
        location: str | None = None
    ) -> None: ...
    bucket_name: str
    object_name: str
    etag: str
    version_id: str | None
    location: str | None

Protocol Types

from typing import Protocol

class ProgressType(Protocol):
    def __call__(self, bytes_amount: int) -> None: ...

class Provider(Protocol):
    def retrieve(self) -> Credentials | None: ...

Version

Tile

Files

tessl/pypi-minio

To install, run

index.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/

MinIO Python SDK

Package Information

Core Imports

Basic Usage

Architecture

Capabilities

Client Operations

Advanced Operations

Administrative Operations

Credentials and Authentication

Configuration Classes

Error Handling

Types

Core Data Types

Protocol Types

index.mddocs/