shanxia/bx/bx-backend/app/services/minio.py

import uuid
from datetime import datetime, timedelta
from io import BytesIO

from minio import Minio

from app.config import get_settings


class MinioService:
    _client: Minio | None = None
    _bucket: str = ""

    @staticmethod
    def _get_client() -> tuple[Minio, str]:
        if MinioService._client is None:
            settings = get_settings()
            MinioService._client = Minio(
                endpoint=settings.MINIO_ENDPOINT,
                access_key=settings.MINIO_ACCESS_KEY,
                secret_key=settings.MINIO_SECRET_KEY,
                secure=settings.MINIO_SECURE,
            )
            MinioService._bucket = settings.MINIO_BUCKET
            if not MinioService._client.bucket_exists(MinioService._bucket):
                MinioService._client.make_bucket(MinioService._bucket)
        return MinioService._client, MinioService._bucket

    @staticmethod
    def upload_file(file_name: str, file_data: bytes, content_type: str) -> str:
        client, bucket = MinioService._get_client()
        ext = file_name.rsplit(".", 1)[-1] if "." in file_name else "bin"
        date_str = datetime.now().strftime("%Y-%m-%d")
        object_name = f"bx/{date_str}/{uuid.uuid4().hex}.{ext}"
        data_stream = BytesIO(file_data)
        client.put_object(
            bucket,
            object_name,
            data_stream,
            length=len(file_data),
            content_type=content_type,
            part_size=10 * 1024 * 1024,
        )
        return object_name

    @staticmethod
    def download_file(file_path: str) -> bytes:
        client, bucket = MinioService._get_client()
        response = client.get_object(bucket, file_path)
        data = response.read()
        response.close()
        response.release_conn()
        return data

    @staticmethod
    def get_presigned_url(file_path: str, expires: int = 3600) -> str:
        client, bucket = MinioService._get_client()
        return client.presigned_get_object(
            bucket, file_path, expires=timedelta(seconds=expires)
        )