Xiaoxia AI
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""MinIO storage service for file uploads."""
|
|
from typing import BinaryIO
|
|
import os
|
|
|
|
from minio import Minio
|
|
from minio.error import S3Error
|
|
|
|
from app.config import get_settings
|
|
|
|
|
|
class MinIOService:
|
|
"""MinIO storage service."""
|
|
|
|
def __init__(self):
|
|
settings = get_settings()
|
|
self.client = Minio(
|
|
settings.MINIO_ENDPOINT,
|
|
access_key=settings.MINIO_ACCESS_KEY,
|
|
secret_key=settings.MINIO_SECRET_KEY,
|
|
secure=settings.MINIO_SECURE,
|
|
)
|
|
self.bucket_name = settings.MINIO_BUCKET
|
|
self.public_url = settings.MINIO_PUBLIC_URL
|
|
self._ensure_bucket()
|
|
|
|
def _ensure_bucket(self):
|
|
try:
|
|
if not self.client.bucket_exists(self.bucket_name):
|
|
self.client.make_bucket(self.bucket_name)
|
|
policy = {
|
|
"Version": "2012-10-17",
|
|
"Statement": [
|
|
{
|
|
"Effect": "Allow",
|
|
"Principal": {"AWS": "*"},
|
|
"Action": ["s3:GetObject"],
|
|
"Resource": [f"arn:aws:s3:::{self.bucket_name}/*"],
|
|
}
|
|
],
|
|
}
|
|
import json
|
|
self.client.set_bucket_policy(self.bucket_name, json.dumps(policy))
|
|
except S3Error as error:
|
|
print(f"Error ensuring bucket: {error}")
|
|
|
|
def upload_file(
|
|
self,
|
|
file: BinaryIO,
|
|
storage_key: str,
|
|
content_type: str = "application/octet-stream",
|
|
) -> str:
|
|
try:
|
|
file.seek(0, os.SEEK_END)
|
|
file_size = file.tell()
|
|
file.seek(0)
|
|
self.client.put_object(
|
|
self.bucket_name,
|
|
storage_key,
|
|
file,
|
|
file_size,
|
|
content_type=content_type,
|
|
)
|
|
return f"{self.public_url}/{self.bucket_name}/{storage_key}"
|
|
except S3Error as error:
|
|
raise Exception(f"Failed to upload file: {error}")
|
|
|
|
def get_url(self, storage_key: str) -> str:
|
|
return f"{self.public_url}/{self.bucket_name}/{storage_key}"
|
|
|
|
def delete_file(self, storage_key: str):
|
|
try:
|
|
self.client.remove_object(self.bucket_name, storage_key)
|
|
except S3Error as error:
|
|
print(f"Error deleting file: {error}")
|
|
|
|
|
|
_minio_service = None
|
|
|
|
|
|
def get_minio_service() -> MinIOService:
|
|
global _minio_service
|
|
if _minio_service is None:
|
|
_minio_service = MinIOService()
|
|
return _minio_service
|