Felipe Cardoso
b46533b18f
The `operation_id` field was added to the `generate_presigned_url` and `upload_file` endpoints. This enhances the OpenAPI documentation by providing unique identifiers for better API clarity and client generation.
106 lines
3.6 KiB
Python
106 lines
3.6 KiB
Python
# app/api/v1/uploads/router.py (refactored version)
|
|
from typing import Dict
|
|
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, status
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.api.dependencies.common import get_storage_provider
|
|
from app.api.dependencies.auth import get_current_user
|
|
from app.core.storage import StorageProvider
|
|
from app.utils.security import verify_upload_token
|
|
from app.utils.files import generate_unique_filename, get_relative_storage_path, validate_image_content_type
|
|
from app.models import User
|
|
from app.schemas.presigned_urls import PresignedUrlResponse, PresignedUrlRequest
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
|
|
@router.post("/presigned-url", response_model=PresignedUrlResponse, operation_id="generate_presigned_url")
|
|
async def generate_presigned_url(
|
|
request: PresignedUrlRequest,
|
|
current_user: User = Depends(get_current_user),
|
|
storage: StorageProvider = Depends(get_storage_provider)
|
|
):
|
|
"""
|
|
Generate a presigned URL for uploading a file.
|
|
|
|
This endpoint creates a secure token that allows direct upload to the storage system.
|
|
After successful upload, the file will be accessible at the returned file_url.
|
|
"""
|
|
if current_user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid authentication credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
# Validate content type
|
|
if not validate_image_content_type(request.content_type):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Content type {request.content_type} is not allowed"
|
|
)
|
|
|
|
# Generate a unique filename
|
|
unique_filename = generate_unique_filename(request.filename)
|
|
|
|
# Create path relative to the storage root
|
|
relative_path = get_relative_storage_path(request.folder, unique_filename)
|
|
|
|
# Generate the presigned URL
|
|
expires_in = 600 # 10 minutes
|
|
upload_url, file_url = storage.generate_presigned_url(
|
|
file_path=relative_path,
|
|
filename=request.filename,
|
|
content_type=request.content_type,
|
|
expires_in=expires_in
|
|
)
|
|
|
|
return PresignedUrlResponse(
|
|
upload_url=upload_url,
|
|
file_url=file_url,
|
|
expires_in=expires_in
|
|
)
|
|
|
|
|
|
@router.post("/{token}", operation_id="upload_file")
|
|
async def upload_file(
|
|
token: str,
|
|
file: UploadFile = File(...),
|
|
storage: StorageProvider = Depends(get_storage_provider),
|
|
):
|
|
"""
|
|
Upload a file using a presigned URL token.
|
|
|
|
This endpoint handles the actual file upload after a presigned URL is generated.
|
|
The token validates the upload permissions and destination.
|
|
"""
|
|
# Verify the token
|
|
payload = verify_upload_token(token)
|
|
if not payload:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired upload token"
|
|
)
|
|
|
|
# Validate the content type
|
|
expected_content_type = payload["content_type"]
|
|
if file.content_type != expected_content_type:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Expected content type {expected_content_type}, got {file.content_type}"
|
|
)
|
|
|
|
# Get the destination path from the token
|
|
destination = payload["path"]
|
|
|
|
try:
|
|
# Save the file
|
|
await storage.save_file(file, destination)
|
|
|
|
# Return the file URL
|
|
return {"file_url": storage.get_file_url(destination)}
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error uploading file: {str(e)}"
|
|
) |