Build 5 — File-Upload Service (S3 + Signed URLs)

# Build 5 — File-Upload Service ## The brief Build a file-upload service. Client requests a "presigned URL" from your API; uploads file directly to S3 (or compatible); your API tracks the upload. Avoids streaming through your backend. ## Why presigned URLs Naive approach: ```python @app.post("/upload") async def upload(file: UploadFile): contents = await file.read() # bottleneck: holds in memory await s3.upload(file.filename, contents) ``` Problems: backend uses bandwidth and memory; slow. Presigned URLs: - Client asks backend for upload URL (with auth check). - Backend generates a signed S3 URL valid for N minutes. - Client PUTs file directly to S3. - Backend records the upload metadata. Backend never touches the file bytes. Scales to GB uploads with zero backend memory. ## Step 1: Setup ```bash uv add fastapi 'uvicorn[standard]' boto3 sqlalchemy ``` For local dev, use MinIO (S3-compatible): ```bash docker run -p 9000:9000 minio/minio server /data ``` ## Step 2: Data model ```python class Upload(Base): __tablename__ = "uploads" id: Mapped[str] = mapped_column(primary_key=True) # uuid user_id: Mapped[int] filename: Mapped[str] content_type: Mapped[str] size: Mapped[int | None] s3_key: Mapped[str] status: Mapped[str] = mapped_column(default="pending") # "pending", "complete", "failed" created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow) completed_at: Mapped[datetime | None] ``` ## Step 3: Presigned URL endpoint ```python import uuid import boto3 from pydantic import BaseModel s3_client = boto3.client( "s3", endpoint_url=settings.s3_endpoint, aws_access_key_id=settings.s3_access_key, aws_secret_access_key=settings.s3_secret_key, ) class UploadRequest(BaseModel): filename: str content_type: str size: int class UploadResponse(BaseModel): upload_id: str upload_url: str # client PUTs to this fields: dict | None # for POST with form @app.post("/uploads/request", response_model=UploadResponse) async def request_upload( payload: UploadRequest, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): # Validate if payload.size > 100 * 1024 * 1024: # 100MB cap raise HTTPException(413, "file too large") upload_id = str(uuid.uuid4()) s3_key = f"uploads/{current_user.id}/{upload_id}/{payload.filename}" # Generate presigned PUT URL (valid for 15 min) url = s3_client.generate_presigned_url( "put_object", Params={ "Bucket": settings.s3_bucket, "Key": s3_key, "ContentType": payload.content_type, }, ExpiresIn=900, ) # Record pending upload upload = Upload( id=upload_id, user_id=current_user.id, filename=payload.filename, content_type=payload.content_type, size=payload.size, s3_key=s3_key, ) db.add(upload); db.commit() return UploadResponse(upload_id=upload_id, upload_url=url, fields=None) ``` Client gets URL; PUTs file directly to S3 (skip backend). ## Step 4: Confirm upload After client uploads, they call back to confirm: ```python @app.post("/uploads/{upload_id}/complete") async def complete_upload( upload_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): upload = db.get(Upload, upload_id) if not upload: raise HTTPException(404) if upload.user_id != current_user.id: raise HTTPException(403) # Verify upload actually happened try: head = s3_client.head_object(Bucket=settings.s3_bucket, Key=upload.s3_key) upload.size = head["ContentLength"] except s3_client.exceptions.NoSuchKey: raise HTTPException(400, "upload not found in storage") upload.status = "complete" upload.completed_at = datetime.utcnow() db.commit() return upload ``` ## Step 5: Download URL (also presigned) ```python @app.get("/uploads/{upload_id}/download-url") async def get_download_url( upload_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): upload = db.get(Upload, upload_id) if not upload or upload.user_id != current_user.id: raise HTTPException(404) url = s3_client.generate_presigned_url( "get_object", Params={"Bucket": settings.s3_bucket, "Key": upload.s3_key}, ExpiresIn=300, # 5 min ) return {"download_url": url} ``` Time-limited URLs. Don't expose S3 keys publicly. ## Step 6: Cleanup stale uploads Some pending uploads never complete (client abandoned). Cleanup job: ```python # tasks.py async def cleanup_stale_uploads(ctx): stale = await db.execute( select(Upload).where( Upload.status == "pending", Upload.created_at < datetime.utcnow() - timedelta(hours=1), ) ) for upload in stale.scalars():