Python & FastAPI: Build Production REST APIs

Why FastAPI?

FastAPI is the fastest-growing Python web framework. It's used at Microsoft, Uber, Netflix, and hundreds of startups. What makes it special:

Performance: on par with Node.js and Go — built on async Python (ASGI)
Auto documentation: Swagger UI and ReDoc generated from your code
Type safety: built-in validation via Pydantic — catch bugs before they ship
Developer experience: auto-completion, inline validation errors, zero boilerplate

If you know Python, you can build a production API in under an hour with FastAPI.

Course Roadmap (Beginner -> Advanced)

This lesson now follows a progressive path so you can study it like a full course.

Phase 1: Beginner Foundations

Python essentials needed for API work
Build first FastAPI app with one endpoint
Request/response models and validation
Basic CRUD with in-memory storage

Phase 2: Intermediate API Engineering

SQLAlchemy async database integration
Service + router separation
Auth with JWT
Testing and error handling

Phase 3: Advanced Production Patterns

Caching and background jobs
Observability and structured logging
Rate limiting and security hardening
Docker deployment and CI checks

CS50-Style Course Design (Week-by-Week)

This section redesigns the lesson into a full course format similar to university-style progression.

Week 0: Functions, Variables, and API Thinking

Topics:

Python functions and return values
variable scope and side effects
pseudocode before implementation
request/response mental model

Short drills:

function refactor exercise
return-vs-print debugging exercise
input validation drill

Problem set:

build 3 utility functions (parse_price, safe_int, normalize_text)
write tests for edge cases

Week 1: Conditionals and Validation Rules

Topics:

branching logic for API constraints
status code decisions (400, 404, 409, 422)
business rule guards

Short drills:

invalid payload scenario handling
branching simplification with guard clauses

Problem set:

design and implement validation for an order creation flow

Week 2: Loops, Collections, and Data Pipelines

Topics:

list/dict transformations
aggregation and summaries
safe iteration over request payloads

Short drills:

quantity aggregation from nested items
transform raw payload to normalized internal schema

Problem set:

build in-memory order aggregation endpoint with filtering

Week 3: Exceptions and Error Architecture

Topics:

Python exception handling patterns
FastAPI exception handlers
consistent error response contracts

Short drills:

convert raw exceptions to API-safe error payloads
add traceable error logs

Problem set:

implement global handlers for validation, HTTP, and unknown exceptions

Week 4: Libraries and Project Structure

Topics:

FastAPI, Pydantic, SQLAlchemy, Alembic roles
layered architecture (routers, services, models, core)
dependency injection patterns

Short drills:

move route logic into service class
add dependency override for testability

Problem set:

scaffold a clean folder structure and migrate one endpoint end-to-end

Week 5: Unit Tests and Integration Tests

Topics:

pytest setup and async test clients
auth fixture patterns
behavior-focused test design

Short drills:

write one passing and one failing test intentionally
debug assertion and response schema mismatch

Problem set:

create a minimum 10-test suite for auth + orders endpoints

Week 6: File I/O and Configuration

Topics:

.env management
secrets and environment-based settings
migration files and deployment config

Short drills:

local vs production config switch
secret rotation simulation

Problem set:

run migration lifecycle (upgrade, downgrade) and document rollback strategy

Week 7: OOP and Service Design

Topics:

service classes for business logic
repository-style isolation for data access
domain modeling principles

Short drills:

extract order state transition rules to domain function
add role checks without duplicating route code

Problem set:

implement OrderService with transition guards and idempotent creation behavior

Week 8: Et Cetera (Production Engineering)

Topics:

background tasks
caching strategy
request tracing and observability
rate limiting and resilience

Short drills:

add request ID middleware
add cache layer for read endpoint

Problem set:

ship a production-readiness pass with logs, metrics-friendly fields, and incident notes

Final Project

Build OrderFlow v2:

secure, tested, documented FastAPI service
role-based permissions, migrations, and Dockerized deployment
reliability enhancements (idempotency + retries + structured logs)

Deliverables:

source code repository
API docs and runbook
test report and architecture note

Real-Time Case Study Used in This Course

All examples use one practical product:

OrderFlow API for e-commerce/internal ops
manages customers, orders, order items, and auth
includes admin-only actions and audit-ready behavior

This gives you realistic design trade-offs instead of toy snippets.

Python Essentials for FastAPI

Before FastAPI, you need to understand these Python concepts.

Type hints

Python

# Python 3.10+
def greet(name: str, age: int) -> str:
    return f"Hello {name}, you are {age}"

from typing import Optional, List, Dict, Union

def get_user(user_id: int, include_orders: bool = False) -> Optional[dict]:
    ...

# Modern union syntax (Python 3.10+)
def process(value: int | str | None) -> str:
    ...

Dataclasses and Pydantic models

Python

from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    id: int
    name: str
    email: str
    created_at: datetime = field(default_factory=datetime.utcnow)

List/dict comprehensions

Python

# List comprehension
squares = [x**2 for x in range(10)]
evens = [n for n in numbers if n % 2 == 0]

# Dict comprehension
email_map = {user.id: user.email for user in users}

# Generator (lazy, memory efficient)
total = sum(item.price * item.qty for item in cart_items)

async / await

Python

import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # simulates I/O
    return {"id": user_id, "name": "Alice"}

async def main():
    # Run concurrently
    users = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )

asyncio.run(main())

Decorators

Python

import functools
import time

def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"{func.__name__} took {time.time() - start:.2f}s")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)

Beginner Module: Build Your First API in 25 Minutes

Start with a minimal but real API shape.

Python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Product(BaseModel):
    id: int
    name: str
    price: float

products: dict[int, Product] = {}

@app.post("/products", response_model=Product)
def create_product(payload: Product):
    if payload.id in products:
        raise HTTPException(status_code=409, detail="Product already exists")
    products[payload.id] = payload
    return payload

@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int):
    product = products.get(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

What this teaches:

path params and response models
conflict (409) and not found (404) handling
validation before DB integration

Project Setup

Bash

# Create virtual environment
python -m venv venv
source venv/bin/activate        # Linux/Mac
venv\Scripts\activate           # Windows

# Install dependencies
pip install fastapi uvicorn[standard] sqlalchemy alembic pydantic-settings
pip install python-jose[cryptography] passlib[bcrypt] httpx

# Create project structure
mkdir -p app/{models,schemas,routers,services,core}
touch app/{__init__,main,database}.py

orderflow/
├── app/
│   ├── main.py           # FastAPI app, startup
│   ├── database.py       # SQLAlchemy engine
│   ├── models/           # SQLAlchemy ORM models
│   ├── schemas/          # Pydantic request/response models
│   ├── routers/          # Route handlers
│   ├── services/         # Business logic
│   └── core/
│       ├── config.py     # Settings from environment
│       ├── security.py   # JWT, hashing
│       └── deps.py       # Dependency injection
├── alembic/              # Database migrations
├── tests/
└── requirements.txt

Day-by-Day Learning Flow (1-Week FastAPI Sprint)

Day 1: Python typing + first FastAPI endpoints
Day 2: Pydantic schemas and validation rules
Day 3: SQLAlchemy async models + DB session dependency
Day 4: Router/service layering + CRUD endpoints
Day 5: JWT login and role-protected routes
Day 6: Tests + global exception handling
Day 7: Docker run + production checklist

FastAPI Application

Python

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import orders, auth, customers
from app.database import create_tables

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await create_tables()
    yield
    # Shutdown (cleanup)

app = FastAPI(
    title="OrderFlow API",
    description="Order management REST API",
    version="1.0.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://learnixo.io", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router,      prefix="/api/v1/auth",      tags=["auth"])
app.include_router(customers.router, prefix="/api/v1/customers",  tags=["customers"])
app.include_router(orders.router,    prefix="/api/v1/orders",     tags=["orders"])

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

Intermediate Module: Service Layer with Business Rules

Avoid putting all logic directly in route handlers.

Python

# app/services/order_service.py
from fastapi import HTTPException

class OrderService:
    def __init__(self, db):
        self.db = db

    async def create(self, payload, user_id: int):
        if not payload.items:
            raise HTTPException(status_code=400, detail="Order requires at least one item")

        # example business rule: max 20 items per order
        item_count = sum(i.quantity for i in payload.items)
        if item_count > 20:
            raise HTTPException(status_code=422, detail="Order item limit exceeded")

        # persist order (omitted for brevity)
        return {"id": 1, "status": "pending", "customer_id": payload.customer_id}

Why this matters in real projects:

easier testing
reusable business constraints
cleaner routers

Pydantic Models (Schemas)

Pydantic validates and serializes data automatically.

Python

# app/schemas/order.py
from pydantic import BaseModel, field_validator, model_validator
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import List

class OrderStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    SHIPPED = "shipped"
    DELIVERED = "delivered"
    CANCELLED = "cancelled"

class OrderItemCreate(BaseModel):
    product_id: int
    quantity: int

    @field_validator("quantity")
    @classmethod
    def quantity_must_be_positive(cls, v: int) -> int:
        if v <= 0:
            raise ValueError("Quantity must be positive")
        return v

class OrderCreate(BaseModel):
    customer_id: int
    items: List[OrderItemCreate]
    notes: str | None = None

    @model_validator(mode="after")
    def items_not_empty(self) -> "OrderCreate":
        if not self.items:
            raise ValueError("Order must have at least one item")
        return self

class OrderItemResponse(BaseModel):
    id: int
    product_id: int
    product_name: str
    quantity: int
    unit_price: Decimal

    model_config = {"from_attributes": True}  # allows ORM model → schema

class OrderResponse(BaseModel):
    id: int
    customer_id: int
    customer_name: str
    status: OrderStatus
    total: Decimal
    items: List[OrderItemResponse]
    created_at: datetime

    model_config = {"from_attributes": True}

class PagedResponse(BaseModel):
    items: List[OrderResponse]
    page: int
    page_size: int
    total: int
    total_pages: int

Real Validation Scenarios You Should Implement

For real APIs, add validators for:

quantity must be > 0
max notes length
immutable fields on update (e.g., created_at)
enum-safe status transitions

Example status transition guard:

Python

ALLOWED_TRANSITIONS = {
    "pending": {"processing", "cancelled"},
    "processing": {"shipped", "cancelled"},
    "shipped": {"delivered"},
    "delivered": set(),
    "cancelled": set(),
}

def validate_transition(current: str, nxt: str) -> None:
    if nxt not in ALLOWED_TRANSITIONS[current]:
        raise ValueError(f"Invalid transition: {current} -> {nxt}")

SQLAlchemy Models

Python

# app/database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from app.core.config import settings

engine = create_async_engine(settings.database_url, echo=settings.debug)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db() -> AsyncSession:  # dependency
    async with AsyncSessionLocal() as session:
        yield session

# app/models/order.py
from sqlalchemy import String, Integer, Numeric, ForeignKey, DateTime, Enum as SAEnum
from sqlalchemy.orm import relationship, Mapped, mapped_column
from datetime import datetime
from app.database import Base

class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    tier: Mapped[str] = mapped_column(String(20), default="free")
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    is_active: Mapped[bool] = mapped_column(default=True)

    orders: Mapped[list["Order"]] = relationship("Order", back_populates="customer")

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    status: Mapped[str] = mapped_column(String(20), default="pending")
    total: Mapped[float] = mapped_column(Numeric(10, 2), default=0)
    notes: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    customer: Mapped["Customer"] = relationship("Customer", back_populates="orders")
    items: Mapped[list["OrderItem"]] = relationship("OrderItem", back_populates="order", cascade="all, delete-orphan")

Performance Tips for Async SQLAlchemy

select only required columns on list endpoints
add pagination on all collection routes
create indexes for frequent filter fields
avoid N+1 with eager loading strategies
use connection pooling defaults appropriate to environment

Router with Dependency Injection

Python

# app/routers/orders.py
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.schemas.order import OrderCreate, OrderResponse, PagedResponse
from app.services.order_service import OrderService
from app.core.deps import get_current_user, require_role
from app.models.user import User

router = APIRouter()

@router.get("/", response_model=PagedResponse)
async def list_orders(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    status: str | None = Query(None),
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    service = OrderService(db)
    return await service.get_paged(page, page_size, status, current_user.id)

@router.get("/{order_id}", response_model=OrderResponse)
async def get_order(
    order_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    service = OrderService(db)
    order = await service.get_by_id(order_id, current_user.id)
    if not order:
        raise HTTPException(status_code=404, detail=f"Order {order_id} not found")
    return order

@router.post("/", response_model=OrderResponse, status_code=status.HTTP_201_CREATED)
async def create_order(
    payload: OrderCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    service = OrderService(db)
    return await service.create(payload, current_user.id)

@router.delete("/{order_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_order(
    order_id: int,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(require_role("admin")),
):
    service = OrderService(db)
    deleted = await service.delete(order_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Order not found")

JWT Authentication

Python

# app/core/security.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=60))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.jwt_secret, algorithm="HS256")

def decode_token(token: str) -> dict:
    return jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])

# app/core/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.core.security import decode_token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await get_user_by_id(db, user_id)
    if user is None:
        raise credentials_exception
    return user

def require_role(role: str):
    async def checker(user = Depends(get_current_user)):
        if user.role != role:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return user
    return checker

# app/routers/auth.py
@router.post("/login")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db),
):
    user = await authenticate_user(db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect email or password")

    token = create_access_token({"sub": str(user.id), "role": user.role})
    return {"access_token": token, "token_type": "bearer"}

Security Hardening Checklist (Advanced)

rotate JWT secret regularly
set short token expiry + refresh strategy
hash passwords with strong parameters
enforce CORS allowlist (never * in prod)
add rate limiting on auth and write-heavy endpoints
sanitize error messages to avoid leaking internals

Settings and Configuration

Python

# app/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    # Database
    database_url: str = "postgresql+asyncpg://user:pass@localhost/orderflow"

    # JWT
    jwt_secret: str = "change-me-in-production"
    jwt_algorithm: str = "HS256"
    jwt_expire_minutes: int = 60

    # App
    debug: bool = False
    environment: str = "development"

    # Email
    smtp_host: str = "smtp.gmail.com"
    smtp_port: int = 587
    smtp_user: str = ""
    smtp_password: str = ""

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

settings = Settings()

Bash

# .env (never commit this)
DATABASE_URL=postgresql+asyncpg://postgres:password@localhost/orderflow
JWT_SECRET=super-secret-key-minimum-32-characters
ENVIRONMENT=production
DEBUG=false

Global Error Handling

Python

# app/main.py
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "errors": exc.errors(),
        },
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    import logging
    logging.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )

Observability (Real Production Requirement)

Add request IDs and structured logging so incidents are traceable.

Python

import logging
import uuid
from fastapi import Request

logger = logging.getLogger("orderflow")

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    logger.info("request_complete", extra={
        "request_id": request_id,
        "path": request.url.path,
        "method": request.method,
        "status_code": response.status_code,
    })
    return response

Background Tasks

Python

from fastapi import BackgroundTasks

@router.post("/orders", status_code=201)
async def create_order(
    payload: OrderCreate,
    background_tasks: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    order = await OrderService(db).create(payload, current_user.id)

    # Fire and forget — doesn't delay the response
    background_tasks.add_task(send_confirmation_email, order.id, current_user.email)

    return order

async def send_confirmation_email(order_id: int, email: str):
    # runs after the response is sent
    await email_service.send(email, f"Order #{order_id} confirmed")

Caching Example (High-Traffic Read Endpoint)

Python

from fastapi import APIRouter
from cachetools import TTLCache

router = APIRouter()
order_cache: TTLCache[int, dict] = TTLCache(maxsize=1000, ttl=30)

@router.get("/{order_id}")
async def get_order_cached(order_id: int):
    cached = order_cache.get(order_id)
    if cached:
        return cached
    # fetch from service/db
    order = {"id": order_id, "status": "pending"}  # replace with real fetch
    order_cache[order_id] = order
    return order

This pattern is useful for frequently-read, slowly-changing resources.

Database Migrations with Alembic

Bash

# Initialize Alembic
alembic init alembic

# Create migration
alembic revision --autogenerate -m "create_orders_table"

# Run migrations
alembic upgrade head

# Rollback one step
alembic downgrade -1

Python

# alembic/env.py
from app.database import Base
from app.models import *  # import all models so Alembic finds them

target_metadata = Base.metadata

Testing

Python

# tests/test_orders.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c

@pytest.fixture
async def auth_headers(client):
    response = await client.post("/api/v1/auth/login", data={
        "username": "test@example.com",
        "password": "testpassword",
    })
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

@pytest.mark.asyncio
async def test_create_order(client, auth_headers):
    response = await client.post(
        "/api/v1/orders",
        json={"customer_id": 1, "items": [{"product_id": 1, "quantity": 2}]},
        headers=auth_headers,
    )
    assert response.status_code == 201
    data = response.json()
    assert data["status"] == "pending"

@pytest.mark.asyncio
async def test_get_order_not_found(client, auth_headers):
    response = await client.get("/api/v1/orders/99999", headers=auth_headers)
    assert response.status_code == 404

Testing Matrix (What to Cover)

auth success/failure
input validation failures
forbidden role checks
happy-path CRUD
not-found paths
service-layer business rule violations

Aim for confidence on behavior, not just line coverage percentage.

Running in Production

DOCKERFILE

# Dockerfile
FROM python:3.12-slim
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Bash

# Run locally
uvicorn app.main:app --reload --port 8000

# View auto-generated docs
open http://localhost:8000/docs      # Swagger UI
open http://localhost:8000/redoc     # ReDoc

Production Readiness Rubric

Before shipping, verify:

[ ] all endpoints have response models
[ ] auth and role checks are enforced
[ ] migrations run clean in CI
[ ] tests cover main success and failure paths
[ ] logs are structured and traceable
[ ] secrets are loaded only from environment
[ ] Docker image builds and starts consistently

Advanced Extensions (After This Lesson)

Add Redis caching and invalidation strategy
Add idempotency keys for payment/order creation
Add audit trail and admin activity logs
Add OpenTelemetry tracing and metrics dashboard
Add async worker queue for heavy tasks

Deep Debugging Labs (Real Incident Scenarios)

Use these labs to move from "I can build" to "I can diagnose and fix production issues."

Lab 1: N+1 Query Performance Bug

Symptom:

GET /orders becomes slow as data grows

Task:

reproduce with seed data
inspect SQL query count
fix using eager loading/selectinload
remeasure latency

Success criteria:

query count reduced significantly
p95 endpoint latency improved

Lab 2: Token Decode Failures

Symptom:

random 401 after deployment

Task:

inspect JWT claims (sub, exp, role)
validate timezone/expiry assumptions
enforce consistent token creation and decode logic
add tests for expired/malformed token paths

Lab 3: CORS Misconfiguration

Symptom:

browser clients fail preflight, Postman works

Task:

capture failing OPTIONS request
set explicit origin allowlist
verify credentials/header/method behavior
document safe dev vs prod config

Lab 4: Concurrency and Idempotency

Symptom:

duplicate order creation under retries/network instability

Task:

add Idempotency-Key header support
persist request fingerprint + response
return previous result for repeated key

Architecture Trade-Offs (When to Choose What)

Sync vs Async Endpoints

choose async for I/O-heavy work (DB calls, HTTP calls)
choose sync for simple CPU-light paths if stack is sync
avoid mixing patterns without clear boundary

ORM vs Raw SQL

ORM for productivity and maintainability
raw SQL for heavy reporting/performance hotspots
keep raw SQL isolated in repository/query modules

JWT vs Session Auth

JWT fits distributed APIs/microservices
sessions can simplify revocation and browser-centric apps
choose based on architecture and revocation requirements

Monolith Service Layer vs Domain Modules

start with service layer for speed
split into domain modules when complexity grows
add clear module boundaries before team scale

Operations Playbook (Production Day-2)

Health and Readiness

/health for liveness
/ready for dependency checks (DB/cache/queue)

Structured Logging

include request_id, user_id, route, status_code, latency_ms
never log secrets, tokens, or PII directly

Timeout and Retry Policy

set per-dependency timeout budgets
retry only transient failures
use exponential backoff with jitter

Rate Limiting

protect auth routes and write-heavy endpoints
use tenant/user-aware limits

Incident Checklist

identify impact scope
isolate failing dependency
degrade gracefully (feature flags/fallback)
patch + verify + postmortem

Capstone: OrderFlow v2 (Senior-Level Build)

Build a production-style API version with:

order + customer CRUD
auth + role-based authorization
idempotent order creation
pagination/filter/sort endpoints
audit logging for admin actions
migration-managed schema
test suite + CI checks
Docker image + deployment guide

Suggested structure:

TEXT

app/
  core/
  models/
  schemas/
  repositories/
  services/
  routers/
tests/
alembic/

Capstone Evaluation Rubric (100 points)

Architecture quality (20): clean layering and boundaries
API correctness (20): validation, status codes, error handling
Security (15): auth, authorization, secret hygiene
Reliability (15): idempotency, retries, resilience patterns
Testing depth (15): happy path + edge + failure path coverage
Performance/ops (10): pagination, logging, diagnostics
Documentation (5): README/runbook clarity

Minimum recommended pass benchmark: 75/100.

Suggested End-to-End File Slice (Starter)

If you want to make this lesson runnable quickly, implement this minimal stack:

app/main.py (app + middleware + routers)
app/core/config.py (env-backed settings)
app/core/security.py (JWT/hash)
app/database.py (engine/session)
app/routers/orders.py (HTTP layer)
app/services/order_service.py (business logic)
tests/test_orders.py (integration tests)

This file slice gives a complete vertical path from request to persistence.

What to Learn Next

Python Interview Questions: 100 Q&As from beginner to senior
Docker & Kubernetes: containerise and deploy this API
AI Systems Engineering: add LLM capabilities to your FastAPI service

Why FastAPI?

Course Roadmap (Beginner -> Advanced)

Phase 1: Beginner Foundations

Phase 2: Intermediate API Engineering

Phase 3: Advanced Production Patterns

CS50-Style Course Design (Week-by-Week)

Week 0: Functions, Variables, and API Thinking

Week 1: Conditionals and Validation Rules

Week 2: Loops, Collections, and Data Pipelines

Week 3: Exceptions and Error Architecture

Week 4: Libraries and Project Structure

Week 5: Unit Tests and Integration Tests

Week 6: File I/O and Configuration

Week 7: OOP and Service Design

Week 8: Et Cetera (Production Engineering)

Final Project

Real-Time Case Study Used in This Course

Python Essentials for FastAPI

Type hints

Dataclasses and Pydantic models

List/dict comprehensions

async / await

Decorators

Beginner Module: Build Your First API in 25 Minutes

Project Setup

Day-by-Day Learning Flow (1-Week FastAPI Sprint)

FastAPI Application

Intermediate Module: Service Layer with Business Rules

Pydantic Models (Schemas)

Real Validation Scenarios You Should Implement

SQLAlchemy Models

Performance Tips for Async SQLAlchemy

Router with Dependency Injection

JWT Authentication

Security Hardening Checklist (Advanced)

Settings and Configuration

Global Error Handling

Observability (Real Production Requirement)

Background Tasks

Caching Example (High-Traffic Read Endpoint)

Database Migrations with Alembic

Testing

Testing Matrix (What to Cover)

Running in Production

Production Readiness Rubric

Advanced Extensions (After This Lesson)

Deep Debugging Labs (Real Incident Scenarios)

Lab 1: N+1 Query Performance Bug

Lab 2: Token Decode Failures

Lab 3: CORS Misconfiguration

Lab 4: Concurrency and Idempotency

Architecture Trade-Offs (When to Choose What)

Sync vs Async Endpoints

ORM vs Raw SQL

JWT vs Session Auth

Monolith Service Layer vs Domain Modules

Operations Playbook (Production Day-2)

Health and Readiness

Structured Logging

Timeout and Retry Policy

Rate Limiting

Incident Checklist

Capstone: OrderFlow v2 (Senior-Level Build)

Capstone Evaluation Rubric (100 points)

Suggested End-to-End File Slice (Starter)

What to Learn Next

REST API Knowledge Check

Enjoyed this article?

Leave a comment