Building APIs with FastAPI

Create production-ready REST APIs with FastAPI, Pydantic models, dependency injection, async endpoints, and automatic documentation

Why FastAPI?

FastAPI provides automatic OpenAPI docs, request validation via Pydantic, async support, and dependency injection: all with high performance.

python

from fastapi import FastAPI

app = FastAPI(title="Advanced API", version="1.0.0")

@app.get("/health")
async def health():
    return {"status": "ok"}

Run with: uvicorn main:app --reload

Visit /docs for Swagger UI and /redoc for ReDoc.

Routing

python

from fastapi import FastAPI, Path, Query, HTTPException, status

app = FastAPI()

# Path parameters
@app.get("/items/{item_id}")
async def read_item(item_id: int = Path(..., ge=1, le=1000)):
    return {"item_id": item_id}

# Query parameters
@app.get("/search")
async def search(
    q: str = Query(None, min_length=3, max_length=50),
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100),
):
    return {"query": q, "page": page, "size": size}

# Multiple HTTP methods
@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(payload: dict):
    return {"id": 42, **payload}

@app.put("/items/{item_id}")
async def update_item(item_id: int, payload: dict):
    return {"updated": item_id}

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    return None

Pydantic Models

python

from pydantic import BaseModel, Field, EmailStr, validator
from datetime import datetime
from typing import Optional, List

class Address(BaseModel):
    street: str
    city: str
    zip_code: str = Field(..., pattern=r"^\d{5}(-\d{4})?$")

class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    address: Optional[Address] = None
    tags: List[str] = Field(default_factory=list, max_length=10)

    @validator("name")
    def name_must_be_meaningful(cls, v):
        if v.strip() == "":
            raise ValueError("Name cannot be blank")
        return v.strip()

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    age: int
    created_at: datetime
    tags: List[str]

    class Config:
        from_attributes = True

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    """Create a new user."""
    return UserResponse(
        id=1,
        name=user.name,
        email=user.email,
        age=user.age,
        created_at=datetime.now(),
        tags=user.tags,
    )

ℹ️Note

Pydantic v2 uses Rust-based validation (via pydantic-core), making it significantly faster than v1. Use Field() for constraints and @validator for custom logic.

Dependency Injection

python

from fastapi import Depends, FastAPI, Header, HTTPException
from typing import Optional

app = FastAPI()

# Simple dependency
async def common_params(page: int = 1, size: int = 10):
    return {"page": page, "size": size}

@app.get("/items")
async def list_items(params: dict = Depends(common_params)):
    return params

# Class-based dependency
class Pagination:
    def __init__(self, page: int = 1, size: int = 10):
        self.page = page
        self.size = size

@app.get("/products")
async def list_products(pagination: Pagination = Depends()):
    return {"page": pagination.page, "size": pagination.size}

# Auth dependency
async def verify_token(authorization: Optional[str] = Header(None)):
    if authorization is None:
        raise HTTPException(status_code=401, detail="Missing token")
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid token")
    return authorization.removeprefix("Bearer ")

@app.get("/protected")
async def protected_route(token: str = Depends(verify_token)):
    return {"message": "Access granted", "token": token}

Dependency with yield (DB sessions)

python

from fastapi import Depends, FastAPI
from typing import Generator

class DatabaseSession:
    def __init__(self):
        self.transaction_active = False

    def commit(self):
        self.transaction_active = False

    def rollback(self):
        self.transaction_active = False

    def close(self):
        pass

async def get_db() -> Generator:
    db = DatabaseSession()
    try:
        yield db
    finally:
        db.close()

@app.post("/transaction")
async def transaction(db: DatabaseSession = Depends(get_db)):
    db.transaction_active = True
    # perform work
    db.commit()
    return {"status": "committed"}

✅Success

Dependency injection with yield is the idiomatic way to manage database sessions, request-scoped resources, and cleanup.

Async Endpoints

python

import asyncio
import aiohttp
from fastapi import FastAPI

app = FastAPI()

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:
        return {"url": url, "status": resp.status}

@app.get("/fetch-all")
async def fetch_all():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/3",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, u) for u in urls]
        results = await asyncio.gather(*tasks)
    return results

Error Handling

python

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

class NotFoundError(BaseModel):
    detail: str
    error_code: str

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc), "error_code": "INVALID_INPUT"},
    )

@app.get("/divide/{a}/{b}")
async def divide(a: float, b: float):
    if b == 0:
        raise HTTPException(
            status_code=400,
            detail="Cannot divide by zero",
        )
    return {"result": a / b}

Real-World: Todo API

python

from fastapi import FastAPI, HTTPException, Depends, status
from pydantic import BaseModel, Field
from typing import List, Optional
from uuid import UUID, uuid4

app = FastAPI(title="Todo API")

class TodoCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    completed: bool = False

class TodoUpdate(BaseModel):
    title: Optional[str] = None
    completed: Optional[bool] = None

class Todo(BaseModel):
    id: UUID
    title: str
    completed: bool

    class Config:
        from_attributes = True

db: List[Todo] = []

@app.get("/todos", response_model=List[Todo])
async def list_todos(completed: Optional[bool] = None):
    if completed is None:
        return db
    return [t for t in db if t.completed == completed]

@app.post("/todos", response_model=Todo, status_code=201)
async def create_todo(todo: TodoCreate):
    new_todo = Todo(id=uuid4(), title=todo.title, completed=todo.completed)
    db.append(new_todo)
    return new_todo

@app.get("/todos/{todo_id}", response_model=Todo)
async def get_todo(todo_id: UUID):
    for t in db:
        if t.id == todo_id:
            return t
    raise HTTPException(404, "Todo not found")

@app.put("/todos/{todo_id}", response_model=Todo)
async def update_todo(todo_id: UUID, update: TodoUpdate):
    for t in db:
        if t.id == todo_id:
            if update.title is not None:
                t.title = update.title
            if update.completed is not None:
                t.completed = update.completed
            return t
    raise HTTPException(404, "Todo not found")

@app.delete("/todos/{todo_id}", status_code=204)
async def delete_todo(todo_id: UUID):
    for i, t in enumerate(db):
        if t.id == todo_id:
            db.pop(i)
            return
    raise HTTPException(404, "Todo not found")

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Practice Questions

Create a FastAPI app with CRUD endpoints for a Product resource (name, price, category, in_stock).
What is dependency injection and how does FastAPI implement it? Show a custom dependency for pagination.
Why use Pydantic's BaseModel over plain dicts for request/response schemas?
Implement an async endpoint that fetches data from 5 external APIs concurrently and returns combined results.
Add authentication to a FastAPI app using a dependency that checks an API key header.
How do you customise the OpenAPI schema? Add descriptions, tags, and a custom servers section.
Build a file upload endpoint that accepts images, validates size and type, and returns a URL.
Compare FastAPI's async endpoints vs sync endpoints. When should you use each?
Implement a middleware that measures request duration and adds it as a response header.
Add database integration using Depends with yield for session management in a FastAPI app.

Progress90%

Behavioral Design Patterns

Package Distribution and PyPI Publishing

Tip: use ← and → to switch lessons

Course Content

1. Threading, Multiprocessing, and Concurrency 2. Asynchronous Programming with asyncio 3. Metaclasses and Descriptors 4. Advanced Decorators 5. Performance Optimization, Cython, and Numba 6. Memory Management and Garbage Collection 7. Creational and Structural Design Patterns 8. Behavioral Design Patterns 9. Building APIs with FastAPI 10. Package Distribution and PyPI Publishing