Felipe Cardoso
2993d0942c
Implemented endpoints for generating presigned URLs and handling file uploads. Added corresponding test cases to ensure proper functionality and error handling. Updated the main router to include the new uploads API.
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)
|
|
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}")
|
|
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)}"
|
|
) |