47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
"""S3 uploads with presigned URLs, plus a local-filesystem mock."""
|
|
import os
|
|
import uuid
|
|
|
|
from app.core.config import settings
|
|
|
|
# Must match the StaticFiles mount in main.py (app/../_uploads = backend/app/_uploads)
|
|
LOCAL_DIR = os.path.abspath(
|
|
os.path.join(
|
|
os.path.dirname(__file__),
|
|
"..",
|
|
"..",
|
|
"_uploads",
|
|
)
|
|
)
|
|
|
|
|
|
async def upload_file(content: bytes, filename: str, content_type: str = "application/octet-stream") -> str:
|
|
key = f"{uuid.uuid4().hex}_{filename}"
|
|
if settings.mock_storage or not (settings.aws_access_key_id and settings.aws_secret_access_key):
|
|
os.makedirs(LOCAL_DIR, exist_ok=True)
|
|
path = os.path.join(LOCAL_DIR, key)
|
|
with open(path, "wb") as f:
|
|
f.write(content)
|
|
# served by FastAPI static mount at /uploads
|
|
return f"/uploads/{key}"
|
|
import boto3
|
|
client = boto3.client(
|
|
"s3", region_name=settings.aws_region,
|
|
aws_access_key_id=settings.aws_access_key_id,
|
|
aws_secret_access_key=settings.aws_secret_access_key,
|
|
)
|
|
client.put_object(Bucket=settings.s3_bucket, Key=key, Body=content, ContentType=content_type)
|
|
return presigned_url(key)
|
|
|
|
|
|
def presigned_url(key: str, expires: int = 3600) -> str:
|
|
import boto3
|
|
client = boto3.client(
|
|
"s3", region_name=settings.aws_region,
|
|
aws_access_key_id=settings.aws_access_key_id,
|
|
aws_secret_access_key=settings.aws_secret_access_key,
|
|
)
|
|
return client.generate_presigned_url(
|
|
"get_object", Params={"Bucket": settings.s3_bucket, "Key": key}, ExpiresIn=expires
|
|
)
|