import asyncio import io from minio import Minio from minio.error import S3Error from app.config import settings _client = Minio( settings.minio_endpoint, access_key=settings.minio_access_key, secret_key=settings.minio_secret_key, secure=settings.minio_secure, ) def _ensure_bucket() -> None: if not _client.bucket_exists(settings.minio_bucket): try: _client.make_bucket(settings.minio_bucket) except S3Error as exc: if exc.code not in {"BucketAlreadyOwnedByYou", "BucketAlreadyExists"}: raise def _put(key: str, payload: bytes, content_type: str) -> None: _client.put_object( settings.minio_bucket, key, io.BytesIO(payload), length=len(payload), content_type=content_type, ) def _exists(key: str) -> bool: try: _client.stat_object(settings.minio_bucket, key) except S3Error as exc: if exc.code in {"NoSuchKey", "NoSuchBucket"}: return False raise return True def _get(key: str) -> tuple[bytes, str]: obj = _client.get_object(settings.minio_bucket, key) try: media_type = obj.headers.get("Content-Type", "application/octet-stream") return obj.read(), media_type finally: obj.close() obj.release_conn() async def ensure_bucket() -> None: await asyncio.to_thread(_ensure_bucket) async def put_object(key: str, payload: bytes, content_type: str) -> None: await asyncio.to_thread(_put, key, payload, content_type) async def object_exists(key: str) -> bool: return await asyncio.to_thread(_exists, key) async def get_object(key: str) -> tuple[bytes, str]: return await asyncio.to_thread(_get, key)