FastAPI w praktyce (routery, Pydantic, zależności)#

FastAPI to fundament backendu ZEUS — ponad 300 endpointów REST. W tej lekcji pokazujemy nasze house rules: jak budować routery, walidować dane Pydantikiem i wstrzykiwać zależności.

Routery zamiast jednego pliku#

Endpointy grupujemy w APIRouter per domena i podpinamy w 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")

Standard ProfessNet: prefiks wersji (/api/v1) ustawiamy raz, przy include_router. Endpointy są cienkie — wołają services/.

Pydantic — walidacja na wejściu i wyjściu#

Każde wejście i wyjście ma model Pydantic. Nigdy nie zwracamy gołego dict-a.

# 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

Walidacja dzieje się automatycznie: błędne wejście to czytelny 422, a nie wyjątek 500 w środku logiki.

Dependency Injection#

Depends to mechanizm wstrzykiwania zależności — sesji bazy, użytkownika z tokenu, serwisów. Zależności są testowalne i podmienialne.

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="Nieprawidłowy 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)

Wskazówka: w testach podmieniasz zależność przez app.dependency_overrides[get_db] = fake_db — bez ruszania kodu produkcyjnego.

Błędy zwracamy spójnie#

from fastapi import HTTPException

if probe is None:
    raise HTTPException(status_code=404, detail="Probe nie istnieje")

Sytuacja	Status
Złe dane wejściowe	422 (automatycznie z Pydantic)
Brak autoryzacji	401
Brak uprawnień	403
Zasób nie istnieje	404
Zaakceptowano zadanie async	202

Async od początku do końca#

Handlery są async, sesja bazy jest asynchroniczna (SQLAlchemy 2.0 async), a długie operacje delegujemy do ARQ (status 202) zamiast blokować request.

---

Cienkie routery, modele Pydantic na każdej granicy i Depends do wszystkiego, co zewnętrzne — to przepis na API, które łatwo testować i utrzymywać. Asynchroniczność rozwiniemy w następnej lekcji.