Валидация данных

В этом разделе описывается система валидации данных в Boilerplate с использованием Pydantic и встроенных механизмов FastAPI. Вы узнаете, как проверять входящие данные, создавать пользовательские валидаторы и эффективно обрабатывать ошибки валидации.

Обзор системы валидации

FastAPI использует Pydantic для автоматической валидации, сериализации и документирования данных. Система валидации обеспечивает:

Проверку типов данных
Валидацию значений (минимальные/максимальные значения, регулярные выражения и т.д.)
Автоматическое преобразование типов (когда это возможно)
Проверку взаимозависимых полей
Автоматическую генерацию документации (OpenAPI/Swagger)

Базовая валидация с Pydantic

Основой системы валидации являются Pydantic модели. Вот пример модели с различными типами валидации:

python

# app/schemas/user.py
from typing import List, Optional
from datetime import date, datetime
from pydantic import BaseModel, EmailStr, Field, validator

class UserCreate(BaseModel):
    """
    Схема для создания пользователя с валидацией полей.
    """
    email: EmailStr = Field(..., description="Email пользователя")
    password: str = Field(
        ...,
        min_length=8,
        max_length=100,
        description="Пароль пользователя (от 8 до 100 символов)"
    )
    full_name: str = Field(
        ...,
        min_length=1,
        max_length=100,
        description="Полное имя пользователя"
    )
    birth_date: Optional[date] = Field(
        None,
        description="Дата рождения пользователя"
    )
    is_active: bool = Field(
        True,
        description="Активен ли пользователь"
    )
    tags: List[str] = Field(
        default_factory=list,
        description="Теги, связанные с пользователем"
    )
    
    # Валидатор пароля на сложность
    @validator("password")
    def password_strength(cls, v):
        """
        Проверяет пароль на сложность.
        Пароль должен содержать как минимум одну цифру, одну заглавную и одну строчную букву.
        """
        if not any(c.isdigit() for c in v):
            raise ValueError("Пароль должен содержать хотя бы одну цифру")
        if not any(c.isupper() for c in v):
            raise ValueError("Пароль должен содержать хотя бы одну заглавную букву")
        if not any(c.islower() for c in v):
            raise ValueError("Пароль должен содержать хотя бы одну строчную букву")
        return v
    
    # Валидатор даты рождения
    @validator("birth_date")
    def birth_date_not_in_future(cls, v):
        """
        Проверяет, что дата рождения не в будущем.
        """
        if v and v > date.today():
            raise ValueError("Дата рождения не может быть в будущем")
        return v
    
    # Валидатор тегов
    @validator("tags")
    def tags_unique(cls, v):
        """
        Проверяет, что теги уникальны.
        """
        if len(set(v)) != len(v):
            raise ValueError("Теги должны быть уникальными")
        return v

Валидация входящих запросов в FastAPI

FastAPI автоматически использует Pydantic модели для валидации данных в запросах:

python

# app/api/routes/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

from app import crud, schemas
from app.api import deps

router = APIRouter()

@router.post("/", response_model=schemas.User, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: schemas.UserCreate,  # Данные автоматически валидируются
    db: AsyncSession = Depends(deps.get_db),
):
    """
    Создание нового пользователя.
    """
    # Проверка, существует ли пользователь с таким email
    existing_user = await crud.user.get_by_email(db, email=user_in.email)
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Пользователь с таким email уже существует",
        )
    
    # Создание пользователя
    user = await crud.user.create(db, obj_in=user_in)
    return user

@router.get("/search", response_model=List[schemas.User])
async def search_users(
    # Параметры запроса с валидацией
    name: str = None,
    email: str = None,
    age_gt: int = None,
    is_active: bool = True,
    limit: int = 10,
    offset: int = 0,
    db: AsyncSession = Depends(deps.get_db),
):
    """
    Поиск пользователей с различными фильтрами.
    """
    # Параметры запроса также автоматически валидируются
    if limit > 100:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Максимальное значение limit - 100",
        )
    
    # Поиск пользователей
    users = await crud.user.search(
        db,
        name=name,
        email=email,
        age_gt=age_gt,
        is_active=is_active,
        limit=limit,
        offset=offset,
    )
    return users

FastAPI валидирует:

Тело запроса (request body)
Параметры пути (path parameters)
Параметры запроса (query parameters)
Заголовки (headers)
Cookies
Form-данные
Файлы

Расширенная валидация

Зависимые поля

Pydantic позволяет валидировать связанные поля с помощью валидаторов и методов root_validator:

python

# app/schemas/product.py
from typing import Optional
from decimal import Decimal
from pydantic import BaseModel, Field, root_validator

class ProductCreate(BaseModel):
    """
    Схема для создания продукта.
    """
    name: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=1000)
    price: Decimal = Field(..., gt=0)
    discount_price: Optional[Decimal] = Field(None, ge=0)
    
    # Валидация зависимых полей
    @root_validator
    def check_prices(cls, values):
        """
        Проверяет, что цена со скидкой меньше обычной цены.
        """
        price = values.get("price")
        discount_price = values.get("discount_price")
        
        if price is not None and discount_price is not None:
            if discount_price >= price:
                raise ValueError("Цена со скидкой должна быть меньше обычной цены")
        
        return values

Динамические значения по умолчанию

python

# app/schemas/article.py
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field

class ArticleCreate(BaseModel):
    """
    Схема для создания статьи.
    """
    title: str = Field(..., min_length=3, max_length=200)
    content: str = Field(..., min_length=10)
    tags: List[str] = Field(default_factory=list)
    # Динамическое значение по умолчанию - текущая дата/время
    published_at: datetime = Field(default_factory=datetime.utcnow)
    is_published: bool = Field(False)

Сложная валидация с регулярными выражениями

python

# app/schemas/custom.py
import re
from typing import Optional
from pydantic import BaseModel, Field, validator

class PhoneNumber(BaseModel):
    """
    Схема для номера телефона с валидацией.
    """
    phone: str = Field(..., min_length=10, max_length=15)
    
    @validator("phone")
    def validate_phone(cls, v):
        """
        Проверяет, что номер телефона соответствует формату.
        """
        pattern = r"^+?[0-9]{10,14}$"
        if not re.match(pattern, v):
            raise ValueError(
                "Номер телефона должен содержать от 10 до 14 цифр и может начинаться с +"
            )
        return v

class PostalCode(BaseModel):
    """
    Схема для почтового индекса с валидацией.
    """
    country_code: str = Field(..., min_length=2, max_length=2)
    postal_code: str = Field(..., min_length=3, max_length=10)
    
    @validator("country_code")
    def validate_country_code(cls, v):
        """
        Проверяет, что код страны состоит из двух заглавных букв.
        """
        if not re.match(r"^[A-Z]{2}$", v):
            raise ValueError("Код страны должен состоять из двух заглавных букв")
        return v
    
    @validator("postal_code")
    def validate_postal_code(cls, v, values):
        """
        Проверяет, что почтовый индекс соответствует формату страны.
        """
        country_code = values.get("country_code")
        
        # Для России
        if country_code == "RU":
            if not re.match(r"^d{6}$", v):
                raise ValueError("Почтовый индекс России должен состоять из 6 цифр")
        
        # Для США
        elif country_code == "US":
            if not re.match(r"^d{5}(-d{4})?$", v):
                raise ValueError(
                    "Почтовый индекс США должен состоять из 5 цифр или 5+4 цифр"
                )
        
        return v

Пользовательские типы и валидаторы

Часто используемые валидаторы можно вынести в отдельные классы или функции:

python

# app/core/validators.py
import re
from datetime import date
from typing import Any, Callable, Dict, Optional, TypeVar

from pydantic import validator

T = TypeVar("T")

def not_empty(field_name: str) -> Callable:
    """
    Валидатор для проверки, что строковое поле не пустое после удаления пробелов.
    
    Использование:
    @validator("field_name")
    def validate_field(cls, v):
        return not_empty("field_name")(cls, v)
    """
    def validate(cls: Any, v: str) -> str:
        if isinstance(v, str):
            v = v.strip()
            if not v:
                raise ValueError(f"Поле {field_name} не может быть пустым")
        return v
    return validate

def not_in_future(field_name: str) -> Callable:
    """
    Валидатор для проверки, что дата не в будущем.
    """
    def validate(cls: Any, v: date) -> date:
        if v and v > date.today():
            raise ValueError(f"Поле {field_name} не может быть в будущем")
        return v
    return validate

def password_strength(value: str) -> str:
    """
    Проверяет пароль на соответствие требованиям безопасности.
    """
    if len(value) < 8:
        raise ValueError("Пароль должен содержать не менее 8 символов")
    
    if not re.search(r"[A-Z]", value):
        raise ValueError("Пароль должен содержать хотя бы одну заглавную букву")
    
    if not re.search(r"[a-z]", value):
        raise ValueError("Пароль должен содержать хотя бы одну строчную букву")
    
    if not re.search(r"[0-9]", value):
        raise ValueError("Пароль должен содержать хотя бы одну цифру")
    
    if not re.search(r"[!@#$%^&*(),.?":{}|<>]", value):
        raise ValueError("Пароль должен содержать хотя бы один специальный символ")
    
    return value

# Пример использования:
# from app.core.validators import not_empty, not_in_future, password_strength
#
# class User(BaseModel):
#     username: str
#     birth_date: date
#     password: str
#     
#     _username_not_empty = validator("username")(not_empty("username"))
#     _birth_date_not_in_future = validator("birth_date")(not_in_future("birth_date"))
#     
#     @validator("password")
#     def validate_password(cls, v):
#         return password_strength(v)

Собственные типы данных

python

# app/core/custom_types.py
import re
from pydantic import constr, validator
from typing import Any, Callable, Optional, Type, TypeVar

T = TypeVar("T")

# Кастомный тип для ИНН
INN = constr(regex=r"^[0-9]{10}|[0-9]{12}$")

# Кастомный тип для СНИЛС
SNILS = constr(regex=r"^[0-9]{3}-[0-9]{3}-[0-9]{3} [0-9]{2}$")

# Функция для создания кастомных типов строк
def create_string_type(
    name: str,
    regex: Optional[str] = None,
    min_length: Optional[int] = None,
    max_length: Optional[int] = None,
    validator_fn: Optional[Callable[[str], str]] = None,
) -> Type[str]:
    """
    Создает кастомный тип строки с валидацией.
    """
    namespace = {}
    
    # Базовые ограничения
    if min_length is not None:
        namespace["min_length"] = min_length
    if max_length is not None:
        namespace["max_length"] = max_length
    if regex is not None:
        namespace["regex"] = regex
    
    # Добавляем валидатор, если он предоставлен
    if validator_fn is not None:
        @validator(name, allow_reuse=True)
        def validate(cls: Any, value: str) -> str:
            return validator_fn(value)
        
        namespace["validate"] = validate
    
    # Создаем новый тип
    return type(name, (str,), namespace)

# Примеры кастомных типов
PhoneNumber = create_string_type(
    "PhoneNumber",
    regex=r"^+?[0-9]{10,14}$",
    min_length=10,
    max_length=15,
)

Email = create_string_type(
    "Email",
    regex=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+.[a-zA-Z0-9-.]+$",
)

# Более сложный тип с валидатором
def password_validator(value: str) -> str:
    """
    Валидатор для проверки сложности пароля.
    """
    errors = []
    
    if len(value) < 8:
        errors.append("Пароль должен содержать не менее 8 символов")
    
    if not re.search(r"[A-Z]", value):
        errors.append("Пароль должен содержать хотя бы одну заглавную букву")
    
    if not re.search(r"[a-z]", value):
        errors.append("Пароль должен содержать хотя бы одну строчную букву")
    
    if not re.search(r"[0-9]", value):
        errors.append("Пароль должен содержать хотя бы одну цифру")
    
    if not re.search(r"[!@#$%^&*(),.?":{}|<>]", value):
        errors.append("Пароль должен содержать хотя бы один специальный символ")
    
    if errors:
        raise ValueError(", ".join(errors))
    
    return value

StrongPassword = create_string_type(
    "StrongPassword",
    validator_fn=password_validator,
)

# Использование:
# from app.core.custom_types import INN, PhoneNumber, StrongPassword
#
# class User(BaseModel):
#     inn: Optional[INN] = None
#     phone: PhoneNumber
#     password: StrongPassword

Обработка ошибок валидации

FastAPI автоматически обрабатывает ошибки валидации и возвращает ответ с кодом 422. Вы можете настроить формат ответа с ошибками:

python

# app/core/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from typing import List, Union

async def validation_exception_handler(
    request: Request,
    exc: Union[RequestValidationError, ValidationError],
) -> JSONResponse:
    """
    Обработчик ошибок валидации.
    Форматирует ошибки в более читаемый формат.
    """
    errors = []
    
    for error in exc.errors():
        # Получаем информацию об ошибке
        loc = error.get("loc", [])
        field = loc[-1] if loc else ""
        field_type = loc[0] if loc else ""
        msg = error.get("msg", "")
        
        # Формируем сообщение об ошибке
        if field_type == "body" and len(loc) > 1:
            error_msg = f"Поле '{field}': {msg}"
        elif field_type == "query":
            error_msg = f"Параметр запроса '{field}': {msg}"
        elif field_type == "path":
            error_msg = f"Параметр пути '{field}': {msg}"
        else:
            error_msg = msg
        
        errors.append({
            "field": field,
            "type": field_type,
            "message": error_msg,
            "detail": error,
        })
    
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error": {
                "code": 422,
                "message": "Ошибка валидации данных",
                "errors": errors,
            },
        },
    )

def add_validation_exception_handler(app: FastAPI) -> None:
    """
    Добавляет обработчик ошибок валидации в приложение.
    """
    app.add_exception_handler(RequestValidationError, validation_exception_handler)
    app.add_exception_handler(ValidationError, validation_exception_handler)

Пример ответа с ошибками валидации:

json

{
  "success": false,
  "error": {
    "code": 422,
    "message": "Ошибка валидации данных",
    "errors": [
      {
        "field": "email",
        "type": "body",
        "message": "Поле 'email': значение не является допустимым адресом электронной почты",
        "detail": {
          "loc": ["body", "email"],
          "msg": "значение не является допустимым адресом электронной почты",
          "type": "value_error.email"
        }
      },
      {
        "field": "password",
        "type": "body",
        "message": "Поле 'password': Пароль должен содержать хотя бы одну заглавную букву",
        "detail": {
          "loc": ["body", "password"],
          "msg": "Пароль должен содержать хотя бы одну заглавную букву",
          "type": "value_error"
        }
      }
    ]
  }
}

Валидация с использованием зависимостей

Вы можете создавать функции-зависимости для валидации параметров запроса:

python

# app/api/deps.py
from fastapi import Depends, HTTPException, Path, Query, status
from typing import Optional

async def valid_user_id(
    user_id: int = Path(..., gt=0, description="ID пользователя (положительное число)")
) -> int:
    """
    Проверяет, что ID пользователя является положительным числом.
    """
    return user_id

async def pagination_params(
    page: int = Query(1, gt=0, description="Номер страницы"),
    page_size: int = Query(10, gt=0, le=100, description="Размер страницы (от 1 до 100)"),
) -> dict:
    """
    Валидирует и возвращает параметры пагинации.
    """
    skip = (page - 1) * page_size
    return {"skip": skip, "limit": page_size}

async def search_filter(
    query: Optional[str] = Query(None, min_length=3, max_length=50, description="Поисковый запрос"),
    category: Optional[str] = Query(None, description="Категория для фильтрации"),
    active_only: bool = Query(False, description="Показывать только активные элементы"),
) -> dict:
    """
    Валидирует и возвращает параметры поиска и фильтрации.
    """
    return {
        "query": query,
        "category": category,
        "active_only": active_only,
    }

# Использование в маршрутах:
# @router.get("/users/{user_id}")
# async def get_user(
#     user_id: int = Depends(valid_user_id),
#     pagination: dict = Depends(pagination_params),
#     filters: dict = Depends(search_filter),
#     db: AsyncSession = Depends(get_db),
# ):
#     skip = pagination["skip"]
#     limit = pagination["limit"]
#     query = filters["query"]
#     # ...

Советы и лучшие практики

Используйте различные модели для разных операций (создание, обновление, чтение)
Применяйте иерархию моделей с наследованием для переиспользования валидации
Для часто используемых паттернов валидации создавайте отдельные функции или типы
Валидируйте все входные данные, не полагаясь на клиентскую валидацию
Предоставляйте подробные и понятные сообщения об ошибках
Используйте строгую валидацию для конфиденциальных данных (пароли, финансовая информация)
Тестируйте валидацию, включая граничные случаи и некорректные входные данные

Валидация данных

Обзор системы валидации

Базовая валидация с Pydantic

Валидация входящих запросов в FastAPI

Расширенная валидация

Зависимые поля

Динамические значения по умолчанию

Сложная валидация с регулярными выражениями

Пользовательские типы и валидаторы

Собственные типы данных

Обработка ошибок валидации

Валидация с использованием зависимостей

Советы и лучшие практики

Ресурсы