Урок по треку Python для веба: Django и FastAPI 8 мин

FastAPI и Pydantic: маршруты, Depends, async и lifespan

Маршруты и модели Pydantic v2, Depends, async и sync обработчики, фоновые задачи и lifespan.

FastAPI стал стандартом для новых Python-сервисов и ML-бэкендов. Он построен на Starlette (ASGI) и Pydantic v2: аннотации типов превращаются в валидацию, сериализацию и OpenAPI-документацию. На собесе проверяют, понимаете ли вы, что происходит за декоратором @app.get.

Маршрут и модели

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field, EmailStr

app = FastAPI()

class UserIn(BaseModel):
    email: EmailStr
    name: str = Field(min_length=1, max_length=100)
    age: int | None = Field(default=None, ge=0, le=150)

class UserOut(BaseModel):
    id: int
    email: EmailStr
    name: str

@app.post('/users', response_model=UserOut, status_code=201)
async def create_user(data: UserIn) -> UserOut:
    user = await repo.create(data)
    return user

@app.get('/users')
async def list_users(limit: int = Query(20, le=100), offset: int = 0): ...
  • Параметры пути, query и тело определяются по аннотациям: простые типы идут из пути или query, Pydantic-модель из тела.
  • Невалидные данные дают 422 с подробностями по полям без единой строки кода.
  • response_model (или аннотация возврата) фильтрует ответ: в UserOut нет password_hash, и он не утечёт, даже если вернуть ORM-объект.

Pydantic v2

  • model_validate(data) создаёт модель из словаря или объекта, model_dump() и model_dump_json() сериализуют.
  • @field_validator('name') и @model_validator(mode='after') для своих правил.
  • model_config = ConfigDict(from_attributes=True) позволяет читать ORM-объекты по атрибутам.
  • Ядро написано на Rust (pydantic-core), поэтому v2 значительно быстрее v1. Старый API (.dict(), parse_obj, @validator) устарел.

Dependency Injection: Depends

async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session                      # после ответа сессия закроется

async def current_user(token: str = Depends(oauth2_scheme),
                       session: AsyncSession = Depends(get_session)) -> User: ...

@app.get('/me')
async def me(user: Annotated[User, Depends(current_user)]):
    return user

Зависимости образуют граф, результаты кешируются в пределах одного запроса (одна и та же зависимость вызовется один раз). Зависимости с yield работают как контекстные менеджеры: код после yield выполняется после обработки запроса. В тестах их подменяют через app.dependency_overrides.

async def или def

  • async def выполняется в цикле событий. Внутри можно только неблокирующие вызовы (await асинхронных клиентов). time.sleep, requests, синхронный драйвер БД заблокируют весь процесс.
  • Обычный def FastAPI выполняет в пуле потоков, поэтому синхронный код там безопасен, но число одновременно выполняемых обработчиков ограничено размером пула.

lifespan и фоновые задачи

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient()
    yield
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

lifespan заменил устаревшие @app.on_event('startup'). BackgroundTasks выполняет лёгкую работу после отправки ответа в том же процессе; для тяжёлого и важного (ретраи, гарантия выполнения) берут очередь задач.

Частые ошибки

  • Блокирующие вызовы в async def.
  • Возврат ORM-модели без response_model и утечка полей.
  • Глобальная сессия БД вместо зависимости с yield.
  • BackgroundTasks для задач, которые нельзя потерять при перезапуске.
  • Создание httpx.AsyncClient() на каждый запрос вместо одного на приложение.

Что спросят на собесе

  • Как FastAPI валидирует данные, откуда берётся 422.
  • Как работает Depends, зачем зависимости с yield, как подменить в тестах.
  • Когда писать async def, а когда def.
  • Что делают response_model, lifespan, BackgroundTasks.
  • Что нового в Pydantic v2.

Закрепите за 5 минут

После урока в тренажёре 3-5 проверочных вопросов, дальше модуль уходит в повторения.