Как решалась задача генерации форм с корректной обработкой ошибок валидации на стороне бэкенда?

Как решалась задача генерации форм с корректной обработкой ошибок валидации на стороне бэкенда

Это критичная область — формы есть в каждом приложении, ошибки валидации влияют на UX. Я использовал многоуровневый подход с чёткой иерархией обработки.

1. Архитектура валидации: послойный подход

Presentaton Layer (API)
    ↓ FastAPI validators
Application Layer (Use Cases)
    ↓ Business rules
Domain Layer (Entities)
    ↓ Domain constraints
Infrastructure Layer (DB)
    ↓ Constraints (unique, foreign key)

Правило: Валидируй на КАЖДОМ слое, не полагайся на один.

2. Pydantic для валидации в API

Базовый пример

# app/api/schemas/user.py
from pydantic import BaseModel, Field, EmailStr, field_validator
from typing import Optional
from datetime import datetime

class UserCreateRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: EmailStr  # Автоматическая валидация email
    age: int = Field(..., ge=0, le=150)  # >= 0, <= 150
    phone: Optional[str] = None
    
    @field_validator('name')
    @classmethod
    def name_must_not_be_only_whitespace(cls, v):
        if not v.strip():
            raise ValueError('Name cannot be empty or whitespace')
        return v.strip()
    
    @field_validator('phone')
    @classmethod
    def validate_phone(cls, v):
        if v and not v.isdigit():
            raise ValueError('Phone must contain only digits')
        return v

# Использование
@router.post("/users")
async def create_user(data: UserCreateRequest):
    # data уже валидна!
    return {"message": "User created", "data": data.model_dump()}

3. Кастомные валидаторы для бизнес-логики

# Пример: проверка уникальности email
from pydantic import BaseModel, field_validator
from app.repositories.user_repository import UserRepository

class UserCreateRequest(BaseModel):
    email: str
    
    @field_validator('email')
    @classmethod
    async def validate_email_unique(cls, v):
        # ПРОБЛЕМА: async валидаторы в Pydantic v1 не поддерживаются!
        # Решение: делай проверку в use case
        return v

# Правильный подход: валидация в use case слое
class CreateUserUseCase:
    def __init__(self, repo: UserRepository):
        self.repo = repo
    
    async def execute(self, data: UserCreateRequest) -> User:
        # Проверка уникальности email
        existing_user = await self.repo.get_by_email(data.email)
        if existing_user:
            raise EmailAlreadyExistsError(f"Email {data.email} already registered")
        
        # Создание пользователя
        user = User(
            name=data.name,
            email=data.email,
            age=data.age,
        )
        await self.repo.save(user)
        return user

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

# app/api/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from app.domain.exceptions import (
    EmailAlreadyExistsError,
    UserNotFoundError,
    ValidationFailedError,
)

# Handler для Pydantic ошибок
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
    """Преобразование Pydantic ошибок в user-friendly формат"""
    errors = {}
    
    for error in exc.errors():
        field = error['loc'][0]  # "email", "age", etc.
        message = error['msg']   # "value is not a valid email address"
        
        errors[field] = {
            "message": message,
            "type": error['type'],  # "value_error", "type_error", etc.
        }
    
    return JSONResponse(
        status_code=422,
        content={
            "status": "error",
            "message": "Validation failed",
            "errors": errors,
        },
    )

# Handler для бизнес-логики ошибок
@app.exception_handler(EmailAlreadyExistsError)
async def email_already_exists_handler(request: Request, exc: EmailAlreadyExistsError):
    return JSONResponse(
        status_code=400,
        content={
            "status": "error",
            "message": str(exc),
            "code": "EMAIL_ALREADY_EXISTS",
            "errors": {
                "email": {
                    "message": "Email already registered",
                    "type": "value_error",
                }
            },
        },
    )

5. Стандартизированный формат ошибок

# app/api/responses.py
from pydantic import BaseModel
from typing import Optional, Dict, Any
from enum import Enum

class ErrorType(str, Enum):
    VALIDATION_ERROR = "validation_error"
    BUSINESS_LOGIC_ERROR = "business_logic_error"
    NOT_FOUND_ERROR = "not_found_error"
    CONFLICT_ERROR = "conflict_error"
    SERVER_ERROR = "server_error"

class FieldError(BaseModel):
    message: str
    type: str  # "required", "value_error", "type_error"
    code: Optional[str] = None

class ErrorResponse(BaseModel):
    status: str = "error"
    message: str
    error_type: ErrorType
    errors: Dict[str, FieldError]
    request_id: str  # для отладки

# Пример ответа
"""
{
    "status": "error",
    "message": "Validation failed",
    "error_type": "validation_error",
    "errors": {
        "email": {
            "message": "value is not a valid email address",
            "type": "value_error",
            "code": "invalid_email"
        },
        "age": {
            "message": "ensure this value is greater than 0",
            "type": "value_error",
            "code": "invalid_range"
        }
    },
    "request_id": "req_12345"
}
"""

6. Полный пример: обработка формы регистрации

# app/api/endpoints/auth.py
from fastapi import APIRouter, HTTPException, Depends
from app.api.schemas.user import UserCreateRequest
from app.application.use_cases.create_user import CreateUserUseCase
from app.domain.exceptions import (
    EmailAlreadyExistsError,
    ValidationFailedError,
)
from app.api.dependencies import get_user_repository

router = APIRouter(prefix="/api/v1/auth")

@router.post("/register", responses={
    201: {"description": "User created"},
    422: {"description": "Validation error"},
    400: {"description": "Email already exists"},
})
async def register(
    data: UserCreateRequest,
    repo = Depends(get_user_repository),
):
    try:
        use_case = CreateUserUseCase(repo=repo)
        user = await use_case.execute(data)
        
        return {
            "status": "success",
            "message": "User registered successfully",
            "data": {
                "id": user.id,
                "name": user.name,
                "email": user.email,
            }
        }
    
    except EmailAlreadyExistsError as e:
        raise HTTPException(
            status_code=400,
            detail={
                "status": "error",
                "message": str(e),
                "code": "EMAIL_ALREADY_EXISTS",
                "errors": {
                    "email": {
                        "message": str(e),
                        "type": "value_error",
                    }
                }
            }
        )
    except ValidationFailedError as e:
        raise HTTPException(
            status_code=422,
            detail={
                "status": "error",
                "message": "Validation failed",
                "errors": e.errors,
            }
        )

7. Тестирование валидации

# tests/test_validation.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

class TestUserValidation:
    def test_valid_user_creation(self):
        """Валидные данные должны пройти"""
        response = client.post("/api/v1/auth/register", json={
            "name": "John Doe",
            "email": "john@example.com",
            "age": 30,
        })
        assert response.status_code == 201
        assert response.json()["status"] == "success"
    
    def test_invalid_email(self):
        """Невалидный email должен вернуть ошибку"""
        response = client.post("/api/v1/auth/register", json={
            "name": "John Doe",
            "email": "invalid-email",  # Not a valid email
            "age": 30,
        })
        assert response.status_code == 422
        data = response.json()
        assert "email" in data["errors"]
        assert "not a valid email" in data["errors"]["email"]["message"]
    
    def test_missing_required_field(self):
        """Отсутствие обязательного поля должно вернуть ошибку"""
        response = client.post("/api/v1/auth/register", json={
            "name": "John Doe",
            # email is missing
            "age": 30,
        })
        assert response.status_code == 422
        data = response.json()
        assert "email" in data["errors"]
    
    def test_invalid_age_range(self):
        """Age вне диапазона должен вернуть ошибку"""
        response = client.post("/api/v1/auth/register", json={
            "name": "John Doe",
            "email": "john@example.com",
            "age": 200,  # > 150
        })
        assert response.status_code == 422
        data = response.json()
        assert "age" in data["errors"]
    
    def test_email_already_exists(self):
        """Duplicate email должен вернуть ошибку"""
        # Создаём первого пользователя
        client.post("/api/v1/auth/register", json={
            "name": "John Doe",
            "email": "john@example.com",
            "age": 30,
        })
        
        # Пытаемся создать второго с тем же email
        response = client.post("/api/v1/auth/register", json={
            "name": "Jane Doe",
            "email": "john@example.com",  # duplicate
            "age": 25,
        })
        assert response.status_code == 400
        assert response.json()["code"] == "EMAIL_ALREADY_EXISTS"

8. Кастомные валидаторы с контекстом

from pydantic import BaseModel, field_validator

class PasswordChangeRequest(BaseModel):
    old_password: str = Field(..., min_length=1)
    new_password: str = Field(..., min_length=8)
    confirm_password: str
    
    @field_validator('new_password')
    @classmethod
    def new_password_strong(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError('Password must contain uppercase letter')
        if not any(c.isdigit() for c in v):
            raise ValueError('Password must contain digit')
        return v
    
    @field_validator('confirm_password')
    @classmethod
    def passwords_match(cls, v, values):
        if 'new_password' in values and v != values['new_password']:
            raise ValueError('Passwords do not match')
        return v

9. Интеграция с фронтенд: JSON схема

from pydantic.json_schema import model_json_schema

# Генерация JSON Schema для фронтенда
schema = model_json_schema(UserCreateRequest)

@router.get("/api/v1/forms/user-schema")
async def get_user_form_schema():
    """Фронтенд может использовать для динамического рендеринга форм"""
    return model_json_schema(UserCreateRequest)

"""
Ответ:
{
    "$defs": {
        "UserCreateRequest": {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string",
                    "minLength": 1,
                    "maxLength": 100
                },
                "email": {
                    "type": "string",
                    "format": "email"
                },
                "age": {
                    "type": "integer",
                    "minimum": 0,
                    "maximum": 150
                }
            },
            "required": ["name", "email", "age"]
        }
    }
}
"""

10. Лучшие практики

# 1. Использовать Field для документации
name: str = Field(
    ...,
    min_length=1,
    max_length=100,
    title="Full Name",
    description="User full name",
    examples=["John Doe"],
)

# 2. Группировать related валидаторы
class AddressSchema(BaseModel):
    street: str = Field(..., min_length=1)
    city: str = Field(..., min_length=1)
    country: str = Field(..., min_length=1)
    postal_code: str = Field(..., pattern=r'^\d{5}$')

# 3. Использовать enum для ограниченных значений
from enum import Enum

class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

class UserRequest(BaseModel):
    role: UserRole  # Автоматическая валидация

# 4. Обработать null vs missing
from typing import Optional

class UpdateUserRequest(BaseModel):
    name: Optional[str] = None  # Может быть null
    email: str  # Обязателен

# 5. Не валидируй паросли в plain text!
# Используй хеширование при сохранении
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"])

class CreateUserRequest(BaseModel):
    password: str = Field(..., min_length=8)  # Plain text в запросе

user.password = pwd_context.hash(data.password)  # Hash при сохранении

Ключевые принципы

1. Валидируй на каждом слое — не полагайся на один
2. Четкие error messages — помогай пользователю понять что не так
3. Стандартизированные ответы — все ошибки в одном формате
4. Тестируй граничные случаи — empty strings, wrong types, duplicates
5. Синхронизируй с фронтенд — используй JSON schema
6. Безопасность — никогда не выводи внутренние ошибки БД

Помни: хорошая валидация = половина успеха приложения!