Advanced Features in Fast APIs

What We're Covering

This stage covers the features that take your API from "works locally" to "production ready":

  • CORS — so your frontend can talk to your API
  • Middleware — code that runs on every request
  • Background Tasks — run tasks after sending response
  • File Uploads — handle images and documents
  • APIRouter — organize routes into separate files
  • Custom Exception Handlers — consistent error responses

Part 1 — CORS

What is CORS?

CORS = Cross-Origin Resource Sharing

When your Next.js frontend (running on localhost:3000) tries to call your FastAPI backend (running on localhost:8000) — the browser blocks it by default. Different ports = different origins = CORS error.

You've definitely seen this error before in your NestJS projects:

Access to fetch at 'http://localhost:8000' from origin 'http://localhost:3000' 
has been blocked by CORS policy

FastAPI fixes this with one middleware setup:

# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",      # Next.js dev server
        "http://localhost:3001",
        "https://yourfrontend.com",   # production frontend
    ],
    allow_credentials=True,           # allows cookies and auth headers
    allow_methods=["*"],              # allows GET, POST, PUT, DELETE etc
    allow_headers=["*"],              # allows Authorization, Content-Type etc
)

For development only — allow everything:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],      # never use this in production
    allow_methods=["*"],
    allow_headers=["*"],
)

Always restrict allow_origins in production — only list domains that should access your API.

Part 2 — Middleware

What is Middleware?

Middleware is code that runs on every single request before it reaches your route, and on every response before it goes back to the client.

Think of it like a checkpoint — every request passes through it.

Request → Middleware → Route Handler → Middleware → Response

You use middleware for things that apply to all routes:

  • Logging every request
  • Measuring response time
  • Adding headers to every response
  • Checking API keys

Creating Custom Middleware

# main.py
import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()

    # Code here runs BEFORE the route handler
    print(f"→ {request.method} {request.url}")

    response = await call_next(request)    # call the actual route

    # Code here runs AFTER the route handler
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(round(process_time * 1000, 2)) + "ms"

    print(f"← {response.status_code} ({round(process_time * 1000, 2)}ms)")

    return response

Now every request logs like this in your terminal:

→ GET http://localhost:8000/users
← 200 (3.42ms)

→ POST http://localhost:8000/auth/login
← 200 (145.23ms)

And every response has the processing time in its headers — visible in browser dev tools.

Multiple Middlewares

You can stack multiple middlewares. They run in order:

@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Request: {request.method} {request.url.path}")
    response = await call_next(request)
    return response


@app.middleware("http")
async def add_security_headers(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    return response

Part 3 — Background Tasks

What are Background Tasks?

Sometimes after handling a request you need to do extra work that the client doesn't need to wait for — like sending an email, logging to a file, processing data.

Without background tasks:

Client waits... API sends email (3 seconds)... API returns response
Total wait: 3+ seconds

With background tasks:

Client waits... API returns response immediately
Email sends in background (client already got response)
Total wait: milliseconds

Basic Background Task

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()


def send_welcome_email(email: str, name: str):
    """This runs in background after response is sent."""
    import time
    time.sleep(2)    # simulate email sending delay
    print(f"Email sent to {email}: Welcome {name}!")


@app.post("/users/register")
def register_user(
    name: str,
    email: str,
    background_tasks: BackgroundTasks
):
    # Add task to run after response
    background_tasks.add_task(send_welcome_email, email, name)

    # This response goes back immediately
    # Email sends after this
    return {"message": "Registered successfully! Check your email."}

Client gets response instantly. Email sends 2 seconds later in background. Client never waits.

Background Task with Database

A realistic example — log every login attempt:

from fastapi import FastAPI, BackgroundTasks, Depends
from sqlalchemy.orm import Session
from datetime import datetime
from database import get_db


def log_login_attempt(email: str, success: bool, ip: str):
    """Log to file in background."""
    status = "SUCCESS" if success else "FAILED"
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_entry = f"[{timestamp}] LOGIN {status} — Email: {email} — IP: {ip}\n"

    with open("login_log.txt", "a") as f:
        f.write(log_entry)

    print(log_entry.strip())


@app.post("/auth/login")
def login(
    credentials: schemas.LoginRequest,
    request: Request,
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    user = crud.authenticate_user(db, credentials.email, credentials.password)
    client_ip = request.client.host

    if not user:
        background_tasks.add_task(log_login_attempt, credentials.email, False, client_ip)
        raise HTTPException(status_code=401, detail="Invalid credentials")

    token = auth.create_access_token(data={"user_id": user.id})
    background_tasks.add_task(log_login_attempt, credentials.email, True, client_ip)

    return {"access_token": token, "token_type": "bearer", "user": user}

Multiple Background Tasks

You can add multiple tasks — they all run after response:

@app.post("/orders")
def create_order(
    order: OrderCreate,
    background_tasks: BackgroundTasks,
    current_user = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    new_order = crud.create_order(db, order, current_user.id)

    background_tasks.add_task(send_order_confirmation_email, current_user.email, new_order)
    background_tasks.add_task(notify_warehouse, new_order.id)
    background_tasks.add_task(update_inventory, order.product_id, order.quantity)

    return new_order    # returns immediately, all 3 tasks run after

Part 4 — File Uploads

Handling File Uploads

FastAPI handles file uploads cleanly. You need python-multipart which you already installed.

from fastapi import FastAPI, File, UploadFile

app = FastAPI()


@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": file.size
    }

UploadFile gives you:

  • file.filename — original filename
  • file.content_type — mime type (image/jpeg, application/pdf etc)
  • file.size — file size in bytes
  • await file.read() — file content as bytes

Saving Uploaded File

import os
import uuid
from fastapi import FastAPI, File, UploadFile, HTTPException

app = FastAPI()

UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)    # create folder if not exists

ALLOWED_TYPES = ["image/jpeg", "image/png", "image/webp"]
MAX_SIZE = 5 * 1024 * 1024    # 5MB in bytes


@app.post("/upload/image")
async def upload_image(file: UploadFile = File(...)):
    # Validate file type
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=400,
            detail=f"File type not allowed. Allowed: {ALLOWED_TYPES}"
        )

    # Read file content
    content = await file.read()

    # Validate file size
    if len(content) > MAX_SIZE:
        raise HTTPException(
            status_code=400,
            detail="File too large. Maximum size is 5MB"
        )

    # Generate unique filename to avoid conflicts
    extension = file.filename.split(".")[-1]
    unique_filename = f"{uuid.uuid4()}.{extension}"
    file_path = os.path.join(UPLOAD_DIR, unique_filename)

    # Save file
    with open(file_path, "wb") as f:
        f.write(content)

    return {
        "message": "File uploaded successfully",
        "filename": unique_filename,
        "original_name": file.filename,
        "size": len(content),
        "url": f"/files/{unique_filename}"
    }

Serving Uploaded Files

After saving, you want to serve them via URL:

from fastapi.staticfiles import StaticFiles

# Mount uploads folder as static files
# Now http://localhost:8000/files/image.jpg serves the file
app.mount("/files", StaticFiles(directory="uploads"), name="files")

Upload with Form Data and Fields Together

Sometimes you want to upload a file AND send extra data together:

from fastapi import Form

@app.post("/upload/profile-picture")
async def upload_profile_picture(
    file: UploadFile = File(...),
    user_id: int = Form(...),
    description: str = Form(default="")
):
    # Can't use JSON body with file upload — must use Form fields
    content = await file.read()

    return {
        "user_id": user_id,
        "description": description,
        "filename": file.filename,
        "size": len(content)
    }

Important — when uploading files you cannot use JSON body for other fields. You must use Form() for text fields alongside File().

Multiple File Upload

@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
    results = []

    for file in files:
        content = await file.read()
        extension = file.filename.split(".")[-1]
        unique_name = f"{uuid.uuid4()}.{extension}"
        file_path = os.path.join(UPLOAD_DIR, unique_name)

        with open(file_path, "wb") as f:
            f.write(content)

        results.append({
            "original": file.filename,
            "saved_as": unique_name,
            "size": len(content)
        })

    return {"uploaded": len(results), "files": results}

Part 5 — APIRouter — Organizing Routes

The Problem

As your app grows, main.py becomes thousands of lines with routes for users, posts, products, auth, orders — everything mixed together. Nightmare to maintain.

APIRouter lets you split routes into separate files — exactly like Controllers in NestJS.

Creating Routers

Create a routers/ folder:

fastapi-learning/
├── routers/
│   ├── auth.py
│   ├── users.py
│   ├── posts.py
│   └── uploads.py
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── auth.py
└── dependencies.py

routers/auth.py

# routers/auth.py

from fastapi import APIRouter, HTTPException, Depends, status
from sqlalchemy.orm import Session
import schemas
import crud
import auth as auth_utils
from database import get_db

router = APIRouter(
    prefix="/auth",           # all routes start with /auth
    tags=["Authentication"]   # groups in /docs
)


@router.post("/register", response_model=schemas.UserResponse, status_code=201)
def register(user: schemas.UserCreate, db: Session = Depends(get_db)):
    existing = crud.get_user_by_email(db, user.email)
    if existing:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db, user)


@router.post("/login", response_model=schemas.TokenResponse)
def login(credentials: schemas.LoginRequest, db: Session = Depends(get_db)):
    user = crud.authenticate_user(db, credentials.email, credentials.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid email or password")

    token = auth_utils.create_access_token(data={"user_id": user.id})
    return {"access_token": token, "token_type": "bearer", "user": user}

routers/users.py

# routers/users.py

from fastapi import APIRouter, HTTPException, Depends, status
from sqlalchemy.orm import Session
import schemas
import crud
from database import get_db
from dependencies import get_current_user

router = APIRouter(
    prefix="/users",
    tags=["Users"]
)


@router.get("/me", response_model=schemas.UserResponse)
def get_me(current_user = Depends(get_current_user)):
    return current_user


@router.patch("/me", response_model=schemas.UserResponse)
def update_me(
    user_update: schemas.UserUpdate,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    return crud.update_user(db, current_user.id, user_update)


@router.delete("/me", status_code=status.HTTP_204_NO_CONTENT)
def delete_me(
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    crud.delete_user(db, current_user.id)


@router.get("", response_model=list[schemas.UserResponse])
def get_users(
    skip: int = 0,
    limit: int = 10,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    return crud.get_users(db, skip, limit)


@router.get("/{user_id}", response_model=schemas.UserResponse)
def get_user(
    user_id: int,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    user = crud.get_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

routers/posts.py

# routers/posts.py

from fastapi import APIRouter, HTTPException, Depends, status
from sqlalchemy.orm import Session
import schemas
import crud
from database import get_db
from dependencies import get_current_user

router = APIRouter(
    prefix="/posts",
    tags=["Posts"]
)


@router.post("", response_model=schemas.PostResponse, status_code=201)
def create_post(
    post: schemas.PostCreate,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    return crud.create_post(db, post, author_id=current_user.id)


@router.get("", response_model=list[schemas.PostResponse])
def get_posts(
    skip: int = 0,
    limit: int = 10,
    published: bool | None = None,
    db: Session = Depends(get_db)
):
    return crud.get_posts(db, skip, limit, published)


@router.get("/my", response_model=list[schemas.PostResponse])
def get_my_posts(
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    return crud.get_posts(db, author_id=current_user.id)


@router.get("/{post_id}", response_model=schemas.PostWithAuthor)
def get_post(post_id: int, db: Session = Depends(get_db)):
    post = crud.get_post(db, post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post


@router.patch("/{post_id}", response_model=schemas.PostResponse)
def update_post(
    post_id: int,
    post_update: schemas.PostUpdate,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    post = crud.get_post(db, post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    if post.author_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not authorized")
    return crud.update_post(db, post_id, post_update)


@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(
    post_id: int,
    db: Session = Depends(get_db),
    current_user = Depends(get_current_user)
):
    post = crud.get_post(db, post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    if post.author_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not authorized")
    crud.delete_post(db, post_id)

routers/uploads.py

# routers/uploads.py

import os
import uuid
from fastapi import APIRouter, File, UploadFile, HTTPException, Depends
from fastapi.responses import FileResponse
from dependencies import get_current_user

router = APIRouter(
    prefix="/uploads",
    tags=["Uploads"]
)

UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)
ALLOWED_TYPES = ["image/jpeg", "image/png", "image/webp", "application/pdf"]
MAX_SIZE = 5 * 1024 * 1024


@router.post("/image")
async def upload_image(
    file: UploadFile = File(...),
    current_user = Depends(get_current_user)
):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(status_code=400, detail="File type not allowed")

    content = await file.read()

    if len(content) > MAX_SIZE:
        raise HTTPException(status_code=400, detail="File too large. Max 5MB")

    extension = file.filename.split(".")[-1].lower()
    unique_name = f"{uuid.uuid4()}.{extension}"
    file_path = os.path.join(UPLOAD_DIR, unique_name)

    with open(file_path, "wb") as f:
        f.write(content)

    return {
        "message": "Uploaded successfully",
        "filename": unique_name,
        "url": f"/files/{unique_name}",
        "uploaded_by": current_user.id
    }

Clean main.py — Register All Routers

Now main.py is beautifully clean:

# main.py

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
import time
import models
from database import engine
from routers import auth, users, posts, uploads

# Create database tables
models.Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="Production Ready API",
    description="FastAPI with Auth, Database, File Uploads",
    version="5.0.0"
)

# ── CORS ──────────────────────────────────────────
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ── Middleware ────────────────────────────────────
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = round((time.time() - start) * 1000, 2)
    print(f"{request.method} {request.url.path} → {response.status_code} ({duration}ms)")
    return response

# ── Static Files ──────────────────────────────────
app.mount("/files", StaticFiles(directory="uploads"), name="files")

# ── Routers ───────────────────────────────────────
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(posts.router)
app.include_router(uploads.router)

# ── Root ──────────────────────────────────────────
@app.get("/", tags=["Root"])
def root():
    return {"message": "API is running", "docs": "/docs"}

Open /docs — routes are now organized into groups: Authentication, Users, Posts, Uploads. Much cleaner.

Part 6 — Custom Exception Handlers

Consistent Error Responses

Right now different errors return different formats. Let's make all errors consistent:

# main.py — add these handlers

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from sqlalchemy.exc import SQLAlchemyError


# Handle validation errors (422)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": " → ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })

    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "message": "Validation failed",
            "errors": errors
        }
    )


# Handle database errors (500)
@app.exception_handler(SQLAlchemyError)
async def database_exception_handler(request: Request, exc: SQLAlchemyError):
    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "message": "Database error occurred",
            "detail": str(exc)
        }
    )


# Handle 404 not found
@app.exception_handler(404)
async def not_found_handler(request: Request, exc):
    return JSONResponse(
        status_code=404,
        content={
            "success": False,
            "message": f"Route {request.url.path} not found"
        }
    )

Now all errors have the same shape — frontend can handle them consistently.

Standard Response Wrapper — Optional but Professional

Some teams wrap all responses in a standard format:

# schemas.py — add this

from typing import Generic, TypeVar
from pydantic import BaseModel

T = TypeVar("T")

class APIResponse(BaseModel, Generic[T]):
    success: bool = True
    message: str = "Success"
    data: T | None = None

Use it in routes:

@router.get("/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = crud.get_user(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return APIResponse(data=user, message="User fetched successfully")

Response:

{
    "success": true,
    "message": "User fetched successfully",
    "data": {
        "id": 1,
        "name": "Gagan",
        ...
    }
}

This is a common pattern in production APIs — especially when building for mobile apps.

Final Project Structure

fastapi-learning/
├── routers/
│   ├── __init__.py
│   ├── auth.py
│   ├── users.py
│   ├── posts.py
│   └── uploads.py
├── uploads/               ← uploaded files stored here
├── venv/
├── .env
├── .gitignore
├── main.py
├── database.py
├── models.py
├── schemas.py
├── crud.py
├── auth.py
├── dependencies.py
├── config.py
└── requirements.txt

Save Requirements

pip freeze > requirements.txt

Your requirements.txt will look like:

fastapi==0.115.0
uvicorn==0.30.0
sqlalchemy==2.0.35
passlib[bcrypt]==1.7.4
python-jose[cryptography]==3.3.0
python-multipart==0.0.12
python-dotenv==1.0.1
pydantic-settings==2.5.2

Exercise 🏋️

Add these features to complete the project:

1. Profile Picture Upload:

  • POST /users/me/avatar — upload profile picture (protected)
  • Save filename to user's database record
  • Add avatar_url field to UserResponse
  • Add avatar column to User model

2. Post Cover Image:

  • PATCH /posts/{id}/cover — upload cover image for a post
  • Only post author can upload cover
  • Add cover_image field to Post model and response

3. Request Logging Middleware:

[2025-03-03 10:30:01] POST /auth/login 200 145ms
[2025-03-03 10:30:05] GET /users/me 200 3ms
[2025-03-03 10:30:10] POST /posts 201 12ms

Save logs to requests.log file using background tasks.

4. Pagination Response: Instead of returning plain list, return:

{
    "data": [...],
    "total": 50,
    "page": 1,
    "per_page": 10,
    "total_pages": 5,
    "has_next": true,
    "has_prev": false
}

Create a reusable paginate() helper function in a new utils.py file.


By

No comments:

Post a Comment

Subscribe to: Post Comments (Atom)

Final Real Project in Fast APIs

What We're Building A complete Task Management API — like a mini Trello/Jira backend. This project uses everything from all 6 stages ...

  • Setting up TypeScript in a Project
    Setting up TypeScript in a project is an important first step to leverage its benefits. Here’s a detailed step-by-step guide on how to set i...
  • Primitive Types in TypeScript
    In TypeScript, primitive types are the most basic data types, and they are the building blocks for handling data. They correspond to simple ...
  • What is TypeScript?
    TypeScript is a strongly typed, object-oriented, compiled language developed and maintained by Microsoft. It is a superset of JavaScript, me...