Step 8 — FastAPI in Practice

The single main.py from step 1, the folder layout from step 2 — now we combine them into an API that does not break in practice. Router organization → async → validation → dependencies → errors → rate limiting, all at once.

Router organization — one module, one resource

As the API grows, defining routes directly in main.py hits a limit fast. Split files per resource and group them with APIRouter.

# routers/articles/articles.py
from fastapi import APIRouter

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

@router.get("/")
async def list_articles():
    ...

@router.get("/{article_id}")
async def get_article(article_id: int):
    ...

# main.py
from fastapi import FastAPI
from routers.articles import articles
from routers.health import health

app = FastAPI()
app.include_router(articles.router)
app.include_router(health.router)

routers/{domain}/{resource}.py — one resource per file. main.py only collects include_router calls. A change in one domain does not touch other files.

Async — don't block the event loop

Write endpoints with async def so they run concurrently on the ASGI event loop. The trap is calling synchronous I/O inside an async def.

import requests

# ❌ blocking call inside async — the whole loop stalls
@router.get("/external")
async def call_external():
    return requests.get("https://api.example.com/data").json()

requests.get() holds the thread until the response arrives. Doing this on the event loop stalls every other request with it. Switch to an async client.

import httpx

# ✅ await yields — other requests run in the meantime
@router.get("/external")
async def call_external():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.example.com/data")
        return r.json()

An unavoidable synchronous call (an old library, etc.) is offloaded to a thread pool.

import asyncio

@router.get("/legacy")
async def legacy():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, blocking_fn)
    return result

Validation — Pydantic schemas.py

Do not leave request/response models as inline dicts inside handlers; gather them in one schemas.py. Redefining models per router splits the same schema.

# schemas.py
from pydantic import BaseModel, Field, field_validator

class ArticleIn(BaseModel):
    title: str = Field(..., max_length=120, description="Article title")
    slug: str = Field(..., pattern=r"^[a-z0-9-]+$", description="URL slug")
    body: str = Field(..., min_length=1, description="Body")

    @field_validator("title")
    @classmethod
    def not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Title cannot be blank")
        return v

class ArticleOut(BaseModel):
    id: int
    title: str
    slug: str

The router imports these models and uses them as type hints only.

# routers/articles/articles.py
from schemas import ArticleIn, ArticleOut

@router.post("/", response_model=ArticleOut, status_code=201)
async def create_article(payload: ArticleIn):
    ...

Specifying response_model separates the response serialization schema from the input — it also acts as a guard against leaking secret fields. On validation failure FastAPI automatically returns a 422 stating which field is wrong and why.

Dependency injection — Depends

Logic shared by many endpoints, such as DB connections or auth, is injected with Depends. Do not call psycopg2.connect() directly in each handler.

# deps.py
from db.connection import get_conn

def get_db():
    with get_conn() as conn:
        yield conn

# routers/articles/articles.py
from fastapi import Depends

@router.get("/{article_id}", response_model=ArticleOut)
async def get_article(article_id: int, db = Depends(get_db)):
    with db.cursor() as cur:
        cur.execute("SELECT id, title, slug FROM articles WHERE id = %s", (article_id,))
        row = cur.fetchone()
    ...

Wrapping setup/teardown in a single function via yield is just like a context manager. Within the same request, the same dependency is evaluated only once (cached). For tests, swap in a fake dependency with app.dependency_overrides.

Errors — HTTPException + global handler

Returning a failure as return {"error": "..."} still has status code 200 and an ad-hoc schema. Raise failures as exceptions.

from fastapi import HTTPException

@router.get("/{article_id}", response_model=ArticleOut)
async def get_article(article_id: int, db = Depends(get_db)):
    row = fetch(db, article_id)
    if row is None:
        raise HTTPException(status_code=404, detail="Article not found")
    return row

To respond uniformly even to unexpected exceptions, add a global handler in main.py. Keep the stack trace in the logs only — never expose it in the response body.

# main.py
import logging
from fastapi import Request
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def on_unhandled(request: Request, exc: Exception):
    logger.exception("Unhandled exception")
    return JSONResponse(status_code=500, content={"detail": "Server error"})

Rate limiting + CORS

A rate limit that stops one IP from abusing an endpoint is better declared with a tool's decorator, such as slowapi, than with a hand-written counter.

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.get("/search")
@limiter.limit("10/minute")
async def search(request: Request, q: str):
    ...

The decorator alone does not work — main.py must attach the limiter to the app and register the exceeded-limit handler, or the decorator raises at request time.

# main.py
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

An API called from a browser on another origin explicitly declares its allowed origins via CORSMiddleware. allow_origins=["*"] cannot be combined with credentialed requests and is far too broad — list the domains.

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

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],   # no "*" — explicit list
    allow_methods=["*"],
    allow_headers=["*"],
)

Try it

Build one articles resource as routers/articles/articles.py. Put ArticleIn and ArticleOut in schemas.py, and make GET /articles/{id} raise HTTPException(404) for a missing ID. Register the global exception handler and CORSMiddleware in main.py, then call from /docs and confirm the 422, 404, and 500 responses each.

Deeper

• FastAPI design philosophy
• Python backend folder philosophy

Course wrap-up

We have come through eight steps. Starting from why Python and FastAPI (step 1), we unfolded folders into domains (step 2), connected a PostgreSQL pool (step 3), ran scheduled jobs with APScheduler (step 4), kept crawler ethics (step 5), wired the external → transform → DB pipeline (step 6), confirmed the service was alive with health checks and observability (step 7), and finally arrived at the patterns that keep that API from breaking in practice (step 8).

It all gathers into one line — with a single language, Python, you can build the whole path of receiving data, transforming it, storing it, and exposing that flow reliably. Even starting from a small single file, it grows this far without much cost.

If you are curious about the next journey, take this service to a real cloud in devops-cloud, or meet how to work with AI tooling in ai-agent-tooling. Thank you for coming this far.