92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
from fastapi import FastAPI, HTTPException, Response
|
||
from minio import Minio
|
||
from minio.error import S3Error
|
||
from pydantic import BaseModel
|
||
from datetime import timedelta
|
||
|
||
app = FastAPI()
|
||
|
||
# 初始化 MinIO 客户端
|
||
minio_client = Minio(
|
||
endpoint="192.168.1.28:9500", # MinIO 服务器地址
|
||
access_key="UIiBRvUixj3GkJwvItmy", # Access Key
|
||
secret_key="wgIPqYxaNlDhUds0d3n5v57Yvzg3u9iUXplO7Kdm", # Secret Key
|
||
secure=False # 如果没有使用 HTTPS，则设置为 False
|
||
)
|
||
|
||
# MinIO 配置
|
||
BUCKET_NAME = "test-bucket"
|
||
|
||
|
||
class FileRequest(BaseModel):
|
||
object_name: str # 文件对象名
|
||
expiry_in_seconds: int = 3600 # URL 有效期（默认 1 小时）
|
||
|
||
|
||
@app.post("/download-presigned-url")
|
||
def generate_presigned_url(file_request: FileRequest):
|
||
"""
|
||
生成用于下载文件的预签名 URL
|
||
"""
|
||
try:
|
||
# 检查存储桶是否存在
|
||
if not minio_client.bucket_exists(BUCKET_NAME):
|
||
raise HTTPException(status_code=404, detail=f"Bucket '{BUCKET_NAME}' does not exist.")
|
||
|
||
# 检查对象是否存在
|
||
try:
|
||
minio_client.stat_object(BUCKET_NAME, file_request.object_name)
|
||
except S3Error:
|
||
raise HTTPException(status_code=404, detail=f"Object '{file_request.object_name}' does not exist.")
|
||
|
||
# 将秒数转换为 timedelta
|
||
expiry_timedelta = timedelta(seconds=file_request.expiry_in_seconds)
|
||
|
||
# 生成预签名 URL
|
||
presigned_url = minio_client.presigned_get_object(
|
||
BUCKET_NAME,
|
||
file_request.object_name,
|
||
expires=expiry_timedelta # 传递 timedelta
|
||
)
|
||
return {"presigned_url": presigned_url}
|
||
|
||
except S3Error as e:
|
||
raise HTTPException(status_code=500, detail=f"Error generating presigned URL: {e}")
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Unexpected error: {e}")
|
||
|
||
|
||
@app.get("/download-file/{object_name:path}")
|
||
def download_file(object_name: str):
|
||
"""
|
||
提供文件对象并直接下载
|
||
"""
|
||
try:
|
||
print(f"Requested object: {object_name}")
|
||
|
||
# 检查存储桶是否存在
|
||
if not minio_client.bucket_exists(BUCKET_NAME):
|
||
raise HTTPException(status_code=404, detail=f"Bucket '{BUCKET_NAME}' does not exist.")
|
||
|
||
# 检查对象是否存在
|
||
try:
|
||
stat = minio_client.stat_object(BUCKET_NAME, object_name)
|
||
except S3Error:
|
||
raise HTTPException(status_code=404, detail=f"Object '{object_name}' does not exist.")
|
||
|
||
# 获取文件对象
|
||
response = minio_client.get_object(BUCKET_NAME, object_name)
|
||
|
||
# 设置响应头
|
||
headers = {
|
||
"Content-Disposition": f"attachment; filename*=UTF-8''{object_name.split('/')[-1]}",
|
||
"Content-Type": stat.content_type
|
||
}
|
||
return Response(content=response.read(), headers=headers, media_type=stat.content_type)
|
||
|
||
except S3Error as e:
|
||
raise HTTPException(status_code=500, detail=f"Error downloading file: {e}")
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"Unexpected error: {e}")
|
||
|