52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
"""MinIO 对象存储（基于 SDK）"""
|
||
from app.config import (
|
||
MINIO_ENDPOINT,
|
||
MINIO_PORT,
|
||
MINIO_ACCESS_KEY,
|
||
MINIO_SECRET_KEY,
|
||
MINIO_BUCKET,
|
||
MINIO_USE_SSL,
|
||
MINIO_PUBLIC_URL,
|
||
MINIO_UPLOAD_PREFIX,
|
||
)
|
||
from app.sdk.minio import MinioSDK
|
||
|
||
_minio_sdk: MinioSDK | None = None
|
||
|
||
|
||
def _get_minio_sdk() -> MinioSDK:
|
||
global _minio_sdk
|
||
if _minio_sdk is None:
|
||
_minio_sdk = MinioSDK(
|
||
end_point=MINIO_ENDPOINT,
|
||
access_key=MINIO_ACCESS_KEY,
|
||
secret_key=MINIO_SECRET_KEY,
|
||
bucket=MINIO_BUCKET,
|
||
use_ssl=MINIO_USE_SSL,
|
||
port=MINIO_PORT,
|
||
public_url=MINIO_PUBLIC_URL,
|
||
upload_prefix=MINIO_UPLOAD_PREFIX,
|
||
)
|
||
return _minio_sdk
|
||
|
||
|
||
def get_minio_client():
|
||
"""获取 MinIO 客户端（未配置时返回 None）"""
|
||
sdk = _get_minio_sdk()
|
||
if not sdk.enabled:
|
||
return None
|
||
return sdk.get_client()
|
||
|
||
|
||
def upload_buffer(data: bytes, object_key: str, content_type: str = "application/octet-stream"):
|
||
"""上传文件到 MinIO（未配置时抛出异常）"""
|
||
sdk = _get_minio_sdk()
|
||
if not sdk.enabled:
|
||
raise RuntimeError("MinIO 未配置，请设置 MINIO_ACCESS_KEY、MINIO_SECRET_KEY")
|
||
return sdk.upload_buffer(data, object_key, content_type)
|
||
|
||
|
||
def generate_object_key(ext: str) -> str:
|
||
"""生成上传用的对象键"""
|
||
return _get_minio_sdk().generate_object_key(ext)
|