advanced⏱90 minutesLesson 9 of 10
Building APIs with FastAPI
Create production-ready REST APIs with FastAPI, Pydantic models, dependency injection, async endpoints, and automatic documentation
Building APIs with FastAPI
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 NonePydantic 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 resultsError 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")Practice Questions
- Create a FastAPI app with CRUD endpoints for a
Productresource (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
BaseModelover 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
serverssection. - 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
Dependswithyieldfor session management in a FastAPI app.
Progress90%
Course Content
1. Threading, Multiprocessing, and Concurrency2. Asynchronous Programming with asyncio3. Metaclasses and Descriptors4. Advanced Decorators5. Performance Optimization, Cython, and Numba6. Memory Management and Garbage Collection7. Creational and Structural Design Patterns8. Behavioral Design Patterns9. Building APIs with FastAPI10. Package Distribution and PyPI Publishing