FastAPI in practice (routers, Pydantic, dependencies)#

FastAPI is the foundation of the ZEUS backend — over 300 REST endpoints. In this lesson we show our house rules: how to build routers, validate data with Pydantic and inject dependencies.

Routers instead of one file#

We group endpoints into an APIRouter per domain and wire them up in main.py.

# app/api/inventory.py
from fastapi import APIRouter, Depends, status
from app.schemas.inventory import ScanRequest, ScanResult
from app.services.inventory import run_inventory

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


@router.post("/scan", response_model=ScanResult, status_code=status.HTTP_202_ACCEPTED)
async def scan(req: ScanRequest, svc=Depends(run_inventory)) -> ScanResult:
    return await svc.start(req.forest)

# app/main.py
from fastapi import FastAPI
from app.api import inventory

app = FastAPI(title="ZEUS Inventory")
app.include_router(inventory.router, prefix="/api/v1")

ProfessNet standard: the version prefix (/api/v1) is set once, at include_router. Endpoints are thin — they call services/.

Pydantic — validation on input and output#

Every input and output has a Pydantic model. We never return a bare dict.

# app/schemas/inventory.py
from pydantic import BaseModel, Field


class ScanRequest(BaseModel):
    forest: str = Field(min_length=1, max_length=255)
    deep: bool = False


class ScanResult(BaseModel):
    job_id: str
    forest: str
    enqueued_at: datetime

Validation happens automatically: bad input becomes a readable 422, not a 500 exception in the middle of the logic.

Dependency Injection#

Depends is the mechanism for injecting dependencies — a database session, the user from a token, services. Dependencies are testable and replaceable.

async def get_db() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session


async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    user = await decode_token(token)
    if user is None:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user


@router.get("/probes")
async def list_probes(
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
) -> list[ProbeOut]:
    return await probe_service.list_for(db, user.tenant_id)

Tip: in tests you swap a dependency via app.dependency_overrides[get_db] = fake_db — without touching production code.

We return errors consistently#

from fastapi import HTTPException

if probe is None:
    raise HTTPException(status_code=404, detail="Probe does not exist")

Situation	Status
Bad input data	422 (automatically from Pydantic)
Not authenticated	401
Not authorized	403
Resource does not exist	404
Async job accepted	202

Async from start to finish#

Handlers are async, the database session is asynchronous (SQLAlchemy 2.0 async), and long operations are delegated to ARQ (status 202) instead of blocking the request.

---

Thin routers, Pydantic models at every boundary and Depends for everything external — that's the recipe for an API that's easy to test and maintain. We'll expand on asynchrony in the next lesson.