FastAPI: Мастерство Backend Разработки
Продвинутый Pydantic и DI
Pydantic V2: Строгий контроль данных
Мы уже знаем, как Pydantic помогает определять структуру данных. Теперь давайте сделаем наши модели умнее и строже. В Pydantic V2 конфигурация модели управляется через ConfigDict, который импортируется и используется внутри класса модели.
Представьте, что ваша модель данных — это фейс-контроль в клубе. Она не только проверяет, что у гостя есть билет (правильные типы полей), но и следит, чтобы он не проносил с собой ничего лишнего.
Чтобы запретить передачу полей, не описанных в модели, мы используем параметр extra='forbid' в ConfigDict. Если клиент попытается отправить лишние данные, API вернет ошибку валидации 422. Это защищает наше приложение от неожиданных данных и потенциальных уязвимостей.
from pydantic import BaseModel, ConfigDict
class UserCreate(BaseModel):
# Определяем строгую конфигурацию
model_config = ConfigDict(extra='forbid')
username: str
email: str
# Этот запрос пройдет валидацию:
# {"username": "alex", "email": "alex@example.com"}
# А этот — нет, потому что в нем есть лишнее поле 'age':
# {"username": "alex", "email": "alex@example.com", "age": 30}
Иногда нам нужны поля, которые не приходят от клиента, а вычисляются на основе других данных. Для этого в Pydantic V2 есть декоратор computed_field. Он создает поле «только для чтения», которое будет включено в ответ API, но не ожидается в запросе.
from pydantic import BaseModel, computed_field
class Order(BaseModel):
price: float
quantity: int
@computed_field
@property
def total_cost(self) -> float:
"""Вычисляет общую стоимость заказа."""
return self.price * self.quantity
# Создаем экземпляр
order = Order(price=10.5, quantity=3)
# total_cost вычисляется автоматически
# print(order.model_dump())
# -> {'price': 10.5, 'quantity': 3, 'total_cost': 31.5}
Инварианты домена и сложная валидация
Что, если валидация зависит от нескольких полей сразу? Например, дата окончания кампании не может быть раньше даты начала. Это бизнес-правило, или, как его еще называют, — условие, которое всегда должно быть истинным для вашей модели данных. Для таких проверок используется model_validator.
Декоратор model_validator позволяет написать функцию, которая проверяет всю модель целиком после того, как отдельные поля уже прошли базовую валидацию. Если что-то не так, мы можем вызвать ValueError, и Pydantic превратит его в красивую ошибку валидации.
from datetime import date
from pydantic import BaseModel, model_validator
class MarketingCampaign(BaseModel):
start_date: date
end_date: date
budget: float
@model_validator(mode='after')
def check_dates(self) -> 'MarketingCampaign':
if self.start_date > self.end_date:
raise ValueError('Дата окончания не может быть раньше даты начала')
return self
# Этот объект создастся успешно
campaign_ok = MarketingCampaign(
start_date=date(2024, 1, 1),
end_date=date(2024, 1, 31),
budget=1000
)
# А здесь будет ошибка валидации
# campaign_fail = MarketingCampaign(
# start_date=date(2024, 2, 1),
# end_date=date(2024, 1, 31),
# budget=1000
# )
Продвинутая инъекция зависимостей
Мы уже использовали функции как зависимости. Но что, если наша зависимость имеет состояние или сложную логику инициализации? В таких случаях лучше использовать классы. FastAPI может инжектировать экземпляры классов так же легко, как и результаты функций.
Класс-зависимость — это обычный класс Python. FastAPI создаст его экземпляр и передаст в вашу функцию пути. Это открывает дорогу для реализации таких паттернов, как или Фабрика, прямо в системе DI.
from fastapi import Depends, FastAPI
app = FastAPI()
# Класс, управляющий состоянием (например, кэшем)
class RateLimiter:
def __init__(self, requests_per_minute: int):
self.requests_per_minute = requests_per_minute
# ... здесь могла бы быть логика отслеживания запросов ...
def is_allowed(self, user_id: str) -> bool:
print(f"Проверка лимита для {user_id}: {self.requests_per_minute} в минуту")
return True # Упрощенный пример
# Создаем "фабрику" для нашей зависимости
limiter = RateLimiter(requests_per_minute=100)
@app.get("/items/")
async def read_items(allowed: bool = Depends(limiter.is_allowed)):
if not allowed:
# ... вернуть ошибку 429 Too Many Requests ...
pass
return {"message": "Вот ваши данные"}
Для управления ресурсами, которые нужно открывать и закрывать (например, сессии базы данных или сетевые соединения), FastAPI поддерживает зависимости с yield. Это элегантный способ использовать контекстные менеджеры.
Код до yield выполняется перед обработкой запроса. То, что yield возвращает, инжектируется в эндпоинт. Код после yield (например, в блоке finally) выполняется после того, как ответ отправлен, даже если в процессе произошла ошибка. Это идеальное место для очистки ресурсов.
from typing import Generator
from sqlalchemy.orm import Session
# Предположим, у нас есть настроенный SessionLocal
from .database import SessionLocal
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
# db — это активная сессия SQLAlchemy
user = db.query(User).filter(User.id == user_id).first()
return user
Наконец, зависимости могут зависеть от других зависимостей, выстраиваясь в цепочки. FastAPI автоматически разрешит всю цепочку. Это очень удобно для многоуровневых проверок, например, сначала получить токен, потом по токену получить пользователя, а уже пользователя передать в эндпоинт.
Ключевая особенность такой системы — возможность подменять зависимости во время тестирования. Вместо реальной функции get_db, которая лезет в базу данных, в тестах мы можем подсунуть фейковую, возвращающую mock-сессию. Это делает наши тесты быстрыми, изолированными и надежными.
Как в Pydantic V2 запретить передачу полей, которые не определены в модели, чтобы при этом генерировалась ошибка валидации?
Для какой цели в Pydantic V2 используется декоратор @computed_field?
Теперь вы владеете мощными инструментами для создания надежных и поддерживаемых API. Строгая валидация данных и гибкая система зависимостей — это основа production-ready сервиса на FastAPI.