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

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:

@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

uv add fastapi 'uvicorn[standard]' boto3 sqlalchemy

For local dev, use MinIO (S3-compatible):

docker run -p 9000:9000 minio/minio server /data

Step 2: Data model

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

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:

@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)

@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:

# 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():
        try:
            s3_client.delete_object(Bucket=settings.s3_bucket, Key=upload.s3_key)
        except Exception:
            pass
        upload.status = "failed"
    await db.commit()

# Schedule via ARQ cron

Step 7: Client-side upload

Example JS:

// 1. Request URL
const { upload_id, upload_url } = await fetch("/uploads/request", {
  method: "POST",
  headers: { "Content-Type": "application/json", Authorization: `Bearer ${token}` },
  body: JSON.stringify({ filename: file.name, content_type: file.type, size: file.size }),
}).then(r => r.json());

// 2. PUT to S3 directly
await fetch(upload_url, { method: "PUT", body: file, headers: { "Content-Type": file.type } });

// 3. Confirm
await fetch(`/uploads/${upload_id}/complete`, {
  method: "POST",
  headers: { Authorization: `Bearer ${token}` },
});

Deliverables

• Working presigned-URL upload flow.
• Server tracks pending → complete transitions.
• Download URL endpoint.
• Cleanup job for stale uploads.
• Tests with MinIO or moto (S3 mocking lib).

What you'll learn

• Direct-to-S3 upload pattern (saves backend bandwidth).
• Presigned URL generation with boto3.
• Two-step upload flow (request → upload → complete).
• Stale-upload cleanup.
• Why backend should NOT stream large files.

Stretch

• Multipart upload for >5GB files.
• Image processing on upload (thumbnails via Lambda/ARQ).
• Virus scanning (ClamAV) before marking complete.
• Per-tier storage limits.

---

📊 100-Point Evaluation Rubric (Case 5)

Evaluation Dimension	Weight	Target Standard (Pass ≥ 80%)
1. Presigned Flow	40 pts	Authenticated POST /uploads/request returns time-boxed PUT URL; direct-to-S3 upload works (MinIO locally); status transitions pending→complete.
2. Validation + Tracking	30 pts	Size cap enforced (413), content-type allowlist, per-user key prefixing, completion callback/webhook recorded.
3. Docs + OpenAPI	15 pts	README runs the MinIO flow end-to-end; /docs schemas for request/status responses.
4. Code Quality	15 pts	Backend never buffers file bytes; keys unguessable (uuid4); no credentials in repo.

---

📤 Submission Instructions & Guidelines

1. Deliverable Formats:
   • Option A (Recommended): Upload a single .zip archive containing your project directory: src/, tests/, Dockerfile (where applicable), and README.md with run instructions.
   • Option B: Upload your README.md and include your GitHub repository link in the Submission Notes box below.
2. Pre-Submission Verification:
   • Run your full test suite (pytest) green; include the passing output in your README.
   • Exercise every endpoint once via /docs (Swagger UI) and paste one sample request/response per endpoint into your README.
   • Confirm no secrets are committed (.env git-ignored; secrets via environment only).