MinIO Adapter Tutorial¶

The Minio adapter provides a clean interface for interacting with MinIO and S3-compatible object storage services.

Installation¶

uv add "archipy[minio]"

Configuration¶

Configure the MinIO adapter via environment variables or a MinioConfig object.

Environment Variables¶

MINIO__ENDPOINT=localhost:9000
MINIO__ACCESS_KEY=minioadmin
MINIO__SECRET_KEY=minioadmin
MINIO__SECURE=false
MINIO__REGION=us-east-1

Direct Configuration¶

from archipy.configs.config_template import MinioConfig

config = MinioConfig(
    ENDPOINT="localhost:9000",
    ACCESS_KEY="minioadmin",
    SECRET_KEY="minioadmin",  # noqa: S106
    SECURE=False,
)

Basic Usage¶

Initializing the Adapter¶

from archipy.adapters.minio.adapters import MinioAdapter

# Use global configuration
minio = MinioAdapter()

# Or provide specific configuration
from archipy.configs.config_template import MinioConfig

custom_config = MinioConfig(
    ENDPOINT="play.min.io:9000",
    ACCESS_KEY="your-access-key",
    SECRET_KEY="your-secret-key",
    SECURE=True
)
minio = MinioAdapter(custom_config)

Bucket Operations¶

import logging

# Configure logging
logger = logging.getLogger(__name__)

# Check if bucket exists and create if needed
try:
    if not minio.bucket_exists("my-bucket"):
        # Create bucket
        minio.make_bucket("my-bucket")
except Exception as e:
    logger.error(f"Failed to check/create bucket: {e}")
    raise
else:
    logger.info("Bucket created successfully")

# List all buckets
try:
    buckets = minio.list_buckets()
except Exception as e:
    logger.error(f"Failed to list buckets: {e}")
    raise
else:
    for bucket in buckets:
        logger.info(f"Bucket: {bucket['name']}, Created: {bucket['creation_date']}")

# Remove bucket
try:
    minio.remove_bucket("my-bucket")
except Exception as e:
    logger.error(f"Failed to remove bucket: {e}")
    raise
else:
    logger.info("Bucket removed successfully")

Working with Objects¶

import logging

# Configure logging
logger = logging.getLogger(__name__)

# Upload a file
minio.put_object("my-bucket", "document.pdf", "/path/to/local/document.pdf")

# Download an object
minio.get_object("my-bucket", "document.pdf", "/path/to/download/document.pdf")

# List objects in a bucket
objects = minio.list_objects("my-bucket", prefix="documents/", recursive=True)
for obj in objects:
    logger.info(f"Object: {obj['object_name']}, Size: {obj['size']} bytes")

# Get object metadata
metadata = minio.stat_object("my-bucket", "document.pdf")
logger.info(f"Content type: {metadata['content_type']}")
logger.info(f"Last modified: {metadata['last_modified']}")

# Remove an object
minio.remove_object("my-bucket", "document.pdf")

Generating Presigned URLs¶

import logging

# Configure logging
logger = logging.getLogger(__name__)

# Generate a presigned URL for downloading (valid for 1 hour by default)
download_url = minio.presigned_get_object("my-bucket", "document.pdf")
logger.info(f"Download URL: {download_url}")

# Generate a presigned URL for uploading (with custom expiry time in seconds)
upload_url = minio.presigned_put_object("my-bucket", "new-document.pdf", expires=7200)  # 2 hours
logger.info(f"Upload URL: {upload_url}")

Managing Bucket Policies¶

import logging
import json

# Configure logging
logger = logging.getLogger(__name__)

# Set a read-only policy for a bucket
policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {"AWS": "*"},
            "Action": ["s3:GetObject"],
            "Resource": [f"arn:aws:s3:::my-bucket/*"]
        }
    ]
}
minio.set_bucket_policy("my-bucket", json.dumps(policy))

# Get bucket policy
policy_info = minio.get_bucket_policy("my-bucket")
logger.info(f"Bucket policy: {policy_info['policy']}")

Error Handling¶

The MinioAdapter uses ArchiPy's domain-specific exceptions for consistent error handling:

import logging
from archipy.models.errors import (
    AlreadyExistsError,
    InternalError,
    InvalidArgumentError,
    NotFoundError,
    PermissionDeniedError,
)

# Configure logging
logger = logging.getLogger(__name__)

try:
    minio.make_bucket("existing-bucket")
except AlreadyExistsError as e:
    logger.warning(f"Bucket already exists: {e}")
except PermissionDeniedError as e:
    logger.error(f"Permission denied to create bucket: {e}")
    raise
except InvalidArgumentError as e:
    logger.error(f"Invalid argument: {e}")
    raise
except InternalError as e:
    logger.error(f"Internal error: {e}")
    raise

Performance Optimization¶

The MinioAdapter includes TTL caching for frequently accessed operations:

# Check if bucket exists (cached for 5 minutes)
minio.bucket_exists("my-bucket")

# List buckets (cached for 5 minutes)
minio.list_buckets()

# Clear all caches if needed
minio.clear_all_caches()

Integration with Web Applications¶

FastAPI Example¶

import logging
import os
import tempfile

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import RedirectResponse

from archipy.adapters.minio.adapters import MinioAdapter
from archipy.models.errors import InternalError, NotFoundError, PermissionDeniedError

# Configure logging
logger = logging.getLogger(__name__)

app = FastAPI()
minio = MinioAdapter()

@app.post("/upload/{bucket_name}")
async def upload_file(bucket_name: str, file: UploadFile) -> dict[str, str]:
    try:
        # Save uploaded file to temporary location
        with tempfile.NamedTemporaryFile(delete=False) as temp:
            content = await file.read()
            temp.write(content)
            temp_path = temp.name

        # Upload to MinIO
        try:
            minio.put_object(bucket_name, file.filename, temp_path)
        except InternalError as e:
            logger.error(f"Failed to upload file: {e}")
            raise HTTPException(status_code=500, detail=str(e)) from e
        else:
            logger.info(f"File {file.filename} uploaded to {bucket_name}")
            return {"message": f"File {file.filename} uploaded successfully"}
        finally:
            os.unlink(temp_path)  # Clean up temp file
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Unexpected error during upload: {e}")
        raise HTTPException(status_code=500, detail=str(e)) from e

@app.get("/download/{bucket_name}/{object_name}")
async def download_file(bucket_name: str, object_name: str) -> RedirectResponse:
    try:
        # Generate presigned URL
        url = minio.presigned_get_object(bucket_name, object_name, expires=3600)
    except NotFoundError as e:
        logger.error(f"File not found: {e}")
        raise HTTPException(status_code=404, detail="File not found") from e
    except PermissionDeniedError as e:
        logger.error(f"Permission denied: {e}")
        raise HTTPException(status_code=403, detail="Permission denied") from e
    except InternalError as e:
        logger.error(f"Internal error: {e}")
        raise HTTPException(status_code=500, detail=str(e)) from e
    else:
        logger.info(f"Generated download URL for {object_name}")
        return RedirectResponse(url)