Использование Pydantic в FastAPI для валидации запросов, ответов и ошибок.
Pydantic — сердце валидации данных в FastAPI. Изучим, как создавать надёжные API
FastAPI автоматически использует Pydantic для валидации:
from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
app = FastAPI()
class UserCreate(BaseModel):
username: str = Field(min_length=3, max_length=20)
email: EmailStr
age: int = Field(ge=0, le=150)
class UserResponse(BaseModel):
id: int
username: str
email: str
age: int
@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
# user уже валидирован
return {
"id": 1,
"username": user.username,
"email": user.email,
"age": user.age
}FastAPI автоматически:
from pydantic import BaseModel, Field, EmailStr, FieldValidationInfo
class UserCreate(BaseModel):
"""Модель для создания пользователя"""
username: str = Field(
min_length=3,
max_length=20,
description="Имя пользователя",
examples=["alice_dev"]
)
email: EmailStr
password: str = Field(
min_length=8,
description="Пароль (минимум 8 символов)"
)
password_confirm: str # для валидации совпадения
@model_validator(mode='after')
def check_passwords_match(self) -> 'UserCreate':
if self.password != self.password_confirm:
raise ValueError('Пароли не совпадают')
return selffrom pydantic import BaseModel, computed_field
from datetime import datetime
class UserResponse(BaseModel):
"""Модель для ответа — не включает чувствительные данные"""
id: int
username: str
email: str
created_at: datetime
is_active: bool = True
# password_hash НЕ включается — безопасно
class Config:
from_attributes = Truefrom pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
# Модель для создания
class PostCreate(BaseModel):
title: str = Field(min_length=5, max_length=200)
content: str = Field(min_length=10)
tags: list[str] = Field(default_factory=list)
# Модель для обновления (все поля optional)
class PostUpdate(BaseModel):
title: str | None = Field(min_length=5, max_length=200, default=None)
content: str | None = Field(min_length=10, default=None)
tags: list[str] | None = None
# Модель для ответа
class PostResponse(BaseModel):
id: int
title: str
content: str
tags: list[str]
author_id: int
created_at: datetime
updated_at: datetime | None
class Config:
from_attributes = Truefrom fastapi import FastAPI, Query, Path
from pydantic import BaseModel, Field
app = FastAPI()
class UserFilter(BaseModel):
min_age: int = Field(ge=0, le=150, default=0)
max_age: int = Field(ge=0, le=150, default=150)
is_active: bool = True
@app.get("/users/{user_id}")
async def get_user(
user_id: int = Path(gt=0, description="ID пользователя"),
include_posts: bool = Query(default=False, description="Включить посты"),
limit: int = Query(ge=1, le=100, default=10)
):
return {"user_id": user_id, "limit": limit}
@app.get("/users/search")
async def search_users(filter: UserFilter = Query()):
# Pydantic валидирует query параметры
return {"min_age": filter.min_age, "max_age": filter.max_age}from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class UserFilter(BaseModel):
username: str | None = Field(None, min_length=3)
email_domain: str | None = None
min_age: int | None = Field(None, ge=0)
@app.post("/users/search")
async def search_users(
filter: UserFilter,
limit: int = 10,
offset: int = 0
):
return {
"filter": filter,
"limit": limit,
"offset": offset
}from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError, BaseModel
app = FastAPI()
class UserCreate(BaseModel):
email: str
password: str = Field(min_length=8)
@app.exception_handler(ValidationError)
async def validation_exception_handler(
request: Request,
exc: ValidationError
):
return JSONResponse(
status_code=422,
content={
"error": "validation_error",
"details": exc.errors()
}
)
@app.post("/users")
async def create_user(user: UserCreate):
return userfrom pydantic import BaseModel, Field
class Address(BaseModel):
street: str
city: str
zip_code: str
country: str
class UserCreate(BaseModel):
username: str
email: str
shipping_address: Address
billing_address: Address | None = None
class UserResponse(BaseModel):
id: int
username: str
email: str
shipping_address: Address
class Config:
from_attributes = Truefrom pydantic import BaseModel
from typing import Generic, TypeVar, List
T = TypeVar('T')
class PaginatedResponse(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
per_page: int
pages: int
@computed_field
@property
def has_next(self) -> bool:
return self.page < self.pages
@computed_field
@property
def has_prev(self) -> bool:
return self.page > 1
# Использование в FastAPI
@app.get("/users", response_model=PaginatedResponse[UserResponse])
async def list_users(page: int = 1, per_page: int = 10):
# ... получение данных ...
return PaginatedResponse(
items=users,
total=total,
page=page,
per_page=per_page,
pages=(total + per_page - 1) // per_page
)from fastapi import FastAPI, Depends
from pydantic import BaseModel, Field
app = FastAPI()
class PaginationParams(BaseModel):
page: int = Field(ge=1, default=1)
per_page: int = Field(ge=1, le=100, default=10)
async def get_pagination(
page: int = 1,
per_page: int = 10
) -> PaginationParams:
return PaginationParams(page=page, per_page=per_page)
@app.get("/items")
async def list_items(pagination: PaginationParams = Depends(get_pagination)):
return {
"page": pagination.page,
"per_page": pagination.per_page
}Создайте полный набор моделей для блога:
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime
from typing import Optional, List
# PostCreate: создание поста
# - title: 5-200 символов
# - content: минимум 10 символов
# - tags: список строк
# - is_published: bool = False
# PostUpdate: обновление (все optional)
# PostResponse: ответ
# - id, title, content, tags, is_published
# - author_id, created_at, updated_at
# CommentCreate: создание комментария
# - post_id: int > 0
# - author_email: EmailStr
# - content: 1-1000 символов
# CommentResponse: ответ
# - id, post_id, author_email, content, created_at
# BlogAPI: FastAPI app с endpoints:
# - POST /posts — создание поста
# - GET /posts/{post_id} — получение поста
# - PUT /posts/{post_id} — обновление поста
# - DELETE /posts/{post_id} — удаление поста
# - POST /posts/{post_id}/comments — добавление комментарияДалее: Интеграция с SQLAlchemy: ORM-модели и Pydantic