qiji/backend/app/services/minio_client.py

from datetime import timedelta
from minio import Minio
from app.config import settings

_client: Minio | None = None


def get_minio_client() -> Minio:
    global _client
    if _client is None:
        _client = Minio(
            settings.MINIO_ENDPOINT,
            access_key=settings.MINIO_ACCESS_KEY,
            secret_key=settings.MINIO_SECRET_KEY,
            secure=settings.MINIO_SECURE,
        )
        # Ensure bucket exists
        if not _client.bucket_exists(settings.MINIO_BUCKET):
            _client.make_bucket(settings.MINIO_BUCKET)
    return _client


def generate_presigned_upload_url(object_key: str, expires: int = 600) -> str:
    """Generate a presigned PUT URL for direct upload to MinIO."""
    client = get_minio_client()
    return client.presigned_put_object(settings.MINIO_BUCKET, object_key, expires=timedelta(seconds=expires))


def generate_presigned_download_url(object_key: str, expires: int = 3600) -> str:
    """Generate a presigned GET URL for viewing/downloading an object."""
    client = get_minio_client()
    return client.presigned_get_object(settings.MINIO_BUCKET, object_key, expires=timedelta(seconds=expires))


def delete_objects(object_keys: list[str]) -> None:
    """Delete multiple objects from MinIO."""
    if not object_keys:
        return
    client = get_minio_client()
    from minio.deleteobjects import DeleteObject
    errors = client.remove_objects(settings.MINIO_BUCKET, [DeleteObject(k) for k in object_keys])
    for err in errors:
        pass  # Log errors in production