100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
import logging
|
|
from datetime import timedelta
|
|
|
|
from config.setting import (
|
|
MINIO_ACCESS_KEY,
|
|
MINIO_BUCKET_NAME,
|
|
MINIO_ENDPOINT,
|
|
MINIO_SECRET_KEY,
|
|
)
|
|
from fastapi import UploadFile
|
|
from minio import Minio
|
|
from minio.error import S3Error
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_minio_client():
|
|
"""MinIO 클라이언트를 생성하고 반환합니다."""
|
|
try:
|
|
client = Minio(
|
|
MINIO_ENDPOINT,
|
|
access_key=MINIO_ACCESS_KEY,
|
|
secret_key=MINIO_SECRET_KEY,
|
|
secure=False, # 개발 환경에서는 False, 프로덕션에서는 True 사용
|
|
)
|
|
# 버킷 존재 여부 확인 및 생성
|
|
found = client.bucket_exists(MINIO_BUCKET_NAME)
|
|
if not found:
|
|
client.make_bucket(MINIO_BUCKET_NAME)
|
|
logger.info(f"Bucket '{MINIO_BUCKET_NAME}' created.")
|
|
else:
|
|
logger.info(f"Bucket '{MINIO_BUCKET_NAME}' already exists.")
|
|
return client
|
|
except (S3Error, Exception) as e:
|
|
logger.error(f"Error connecting to MinIO: {e}")
|
|
raise
|
|
|
|
|
|
def upload_file_to_minio(file: UploadFile, bucket_name: str, object_name: str) -> str:
|
|
"""
|
|
파일을 MinIO에 업로드하고, presigned URL을 반환합니다.
|
|
|
|
Args:
|
|
file (UploadFile): FastAPI의 UploadFile 객체
|
|
bucket_name (str): 업로드할 버킷 이름
|
|
object_name (str): 저장될 객체 이름 (경로 포함 가능)
|
|
|
|
Returns:
|
|
str: 생성된 presigned URL
|
|
"""
|
|
minio_client = get_minio_client()
|
|
try:
|
|
# 1. 버킷 존재 확인 및 생성
|
|
found = minio_client.bucket_exists(bucket_name)
|
|
if not found:
|
|
minio_client.make_bucket(bucket_name)
|
|
logger.info(f"✅ 버킷 '{bucket_name}' 생성 완료.")
|
|
|
|
# 2. 파일 업로드
|
|
file.file.seek(0) # 파일 포인터를 처음으로 이동
|
|
minio_client.put_object(
|
|
bucket_name,
|
|
object_name,
|
|
file.file,
|
|
length=-1, # 파일 크기를 모를 때 -1로 설정
|
|
part_size=10 * 1024 * 1024, # 10MB 단위로 청크 업로드
|
|
)
|
|
logger.info(f"✅ '{object_name}' -> '{bucket_name}' 업로드 성공.")
|
|
|
|
# 3. Presigned URL 생성
|
|
presigned_url = minio_client.presigned_get_object(
|
|
bucket_name,
|
|
object_name,
|
|
expires=timedelta(days=7), # URL 만료 기간 (예: 7일, 필요에 따라 조절 가능)
|
|
)
|
|
logger.info(f"✅ Presigned URL 생성 완료: {presigned_url}")
|
|
|
|
return presigned_url
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ MinIO 작업 실패: {e}")
|
|
raise # 실패 시 예외를 다시 발생시켜 호출 측에서 처리하도록 함
|
|
|
|
|
|
def download_file_from_minio(object_name: str, local_path: str):
|
|
"""
|
|
MinIO에서 객체를 다운로드하여 로컬 파일로 저장합니다.
|
|
|
|
Args:
|
|
object_name (str): 다운로드할 객체의 이름
|
|
local_path (str): 파일을 저장할 로컬 경로
|
|
"""
|
|
client = get_minio_client()
|
|
try:
|
|
client.fget_object(MINIO_BUCKET_NAME, object_name, local_path)
|
|
logger.info(f"'{object_name}' downloaded to '{local_path}' successfully.")
|
|
except S3Error as e:
|
|
logger.error(f"Error downloading from MinIO: {e}")
|
|
raise
|