Pydantic в микросервисной архитектуре: schema registry, версионирование и интеграция с FastAPI

from pydantic import BaseModel, Field, model_validator
from typing import Literal
 
# Версия 1: старая схема (deprecated)
class UserV1(BaseModel):
    version: Literal["v1"] = "v1"
    id: int
    name: str  # Одно поле для ФИО
    email: str
 
# Версия 2: новая схема
class UserV2(BaseModel):
    version: Literal["v2"] = "v2"
    id: int
    first_name: str
    last_name: str
    email: str
    phone: str | None = None
 
    @model_validator(mode='before')
    @classmethod
    def migrate_from_v1(cls, data):
        """Автоматическая миграция из v1 при получении старого формата."""
        # Определяем версию по наличию полей
        if 'name' in data and 'first_name' not in data:
            # Это v1 формат — мигрируем
            parts = data['name'].split(' ', 1)
            data['first_name'] = parts[0]
            data['last_name'] = parts[1] if len(parts) > 1 else ''
            data.pop('name', None)
            data['version'] = 'v2'
 
        return data
 
# API endpoint поддерживает обе версии
@app.post("/users")
def create_user(user_data: dict):
    # Определяем версию по данным
    if 'name' in user_data or user_data.get('version') == 'v1':
        user = UserV1(**user_data)
        # Конвертируем в v2 для внутренней логики
        user_v2 = UserV2(
            id=user.id,
            first_name=user.name.split()[0],
            last_name=' '.join(user.name.split()[1:]),
            email=user.email
        )
    else:
        user_v2 = UserV2(**user_data)
 
    # Вся бизнес-логика работает с v2
    db.add(user_v2)
    return user_v2

from fastapi import Header, HTTPException
 
# Разные response модели для разных версий API
class UserResponseV1(BaseModel):
    id: int
    name: str
    email: str
 
class UserResponseV2(BaseModel):
    id: int
    first_name: str
    last_name: str
    email: str
    phone: str | None
 
@app.get("/users/{user_id}")
def get_user(
    user_id: int,
    accept_version: str = Header(default="v2", alias="Accept-Version")
):
    # Получаем пользователя из БД (внутренний формат — v2)
    user = db.query(User).get(user_id)
 
    # Возвращаем в запрошенной версии
    if accept_version == "v1":
        return UserResponseV1(
            id=user.id,
            name=f"{user.first_name} {user.last_name}",
            email=user.email
        )
    elif accept_version == "v2":
        return UserResponseV2(**user.dict())
    else:
        raise HTTPException(400, f"Unsupported API version: {accept_version}")

from pydantic import BaseModel, Field
from datetime import datetime
from uuid import UUID
 
# Экспортируем только v2 (последнюю версию)
class User(BaseModel):
    """User model v2 (current).
 
    Changelog:
    - v2 (2025-01-15): Split name into first_name/last_name, added phone
    - v1 (2024-06-01): Initial version
    """
    id: UUID
    first_name: str = Field(min_length=1, max_length=50)
    last_name: str = Field(min_length=1, max_length=50)
    email: str = Field(pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    phone: str | None = None
    created_at: datetime
    updated_at: datetime | None = None
 
# Для backward compatibility держим v1 (deprecated)
class UserV1(BaseModel):
    """User model v1 (deprecated, will be removed in v3.0.0)."""
    id: UUID
    name: str
    email: str
    created_at: datetime
 
# Версионирование библиотеки через setup.py
# version='1.2.3'
# - Major version (1): breaking changes
# - Minor version (2): new features, backward compatible
# - Patch version (3): bug fixes

# CHANGELOG.md
 
## [2.0.0] - 2025-03-01
### Breaking Changes
- User: removed `name` field, replaced with `first_name` + `last_name`
- Order: `status` now enum instead of string
 
### Migration Guide
Update `user-service` to v2.0.0:
1. Update imports: `from my_company_schemas.user import User`
2. Migrate data: `user.first_name = user.name.split()[0]`
3. Update API clients in dependent services
 
## [1.2.3] - 2025-01-15
### Added
- User: optional `phone` field
 
### Fixed
- Payment: currency validation regex
 
## [1.0.0] - 2024-06-01
### Added
- Initial release with User, Order, Payment models

# schema_registry/models.py
from pydantic import BaseModel
from datetime import datetime
from enum import Enum
 
class SchemaFormat(str, Enum):
    PYDANTIC = "pydantic"
    JSON_SCHEMA = "json_schema"
    AVRO = "avro"
 
class SchemaVersion(BaseModel):
    subject: str  # "user", "order", etc.
    version: int  # Автоинкремент
    schema: dict  # JSON Schema или Pydantic model as dict
    format: SchemaFormat
    created_at: datetime
    created_by: str  # Service name
    is_deprecated: bool = False
 
# schema_registry/service.py
from fastapi import FastAPI, HTTPException
from sqlalchemy.orm import Session
 
app = FastAPI()
 
@app.post("/schemas/{subject}")
def register_schema(
    subject: str,
    schema_data: dict,
    format: SchemaFormat = SchemaFormat.PYDANTIC,
    db: Session = Depends(get_db)
):
    """Регистрация новой версии схемы."""
    # Получаем последнюю версию
    last_version = db.query(SchemaVersion).filter_by(
        subject=subject
    ).order_by(SchemaVersion.version.desc()).first()
 
    # Проверяем обратную совместимость
    if last_version:
        if not is_backward_compatible(last_version.schema, schema_data):
            raise HTTPException(
                400,
                "Schema is not backward compatible. "
                "Breaking changes require major version bump."
            )
 
    # Создаём новую версию
    new_version = SchemaVersion(
        subject=subject,
        version=(last_version.version + 1) if last_version else 1,
        schema=schema_data,
        format=format,
        created_at=datetime.utcnow(),
        created_by=request.headers.get('X-Service-Name', 'unknown')
    )
    db.add(new_version)
    db.commit()
 
    return {"subject": subject, "version": new_version.version}
 
@app.get("/schemas/{subject}/versions/{version}")
def get_schema(subject: str, version: int, db: Session = Depends(get_db)):
    """Получение схемы конкретной версии."""
    schema_version = db.query(SchemaVersion).filter_by(
        subject=subject, version=version
    ).first()
 
    if not schema_version:
        raise HTTPException(404, f"Schema {subject} v{version} not found")
 
    return schema_version
 
@app.get("/schemas/{subject}/versions/latest")
def get_latest_schema(subject: str, db: Session = Depends(get_db)):
    """Получение последней версии схемы."""
    schema_version = db.query(SchemaVersion).filter_by(
        subject=subject
    ).order_by(SchemaVersion.version.desc()).first()
 
    if not schema_version:
        raise HTTPException(404, f"Schema {subject} not found")
 
    return schema_version
 
def is_backward_compatible(old_schema: dict, new_schema: dict) -> bool:
    """Проверка обратной совместимости схем."""
    # Проверяем, что новые required поля не добавлены
    old_required = set(old_schema.get('required', []))
    new_required = set(new_schema.get('required', []))
 
    if not old_required.issubset(new_required):
        return False  # Удалены required поля
 
    # Проверяем, что типы полей не изменились несовместимым образом
    old_props = old_schema.get('properties', {})
    new_props = new_schema.get('properties', {})
 
    for field_name, old_prop in old_props.items():
        if field_name in new_props:
            if old_prop.get('type') != new_props[field_name].get('type'):
                return False  # Изменился тип поля
 
    return True

# my_company_schemas/registry_client.py
import requests
from pydantic import BaseModel, TypeAdapter
from functools import lru_cache
 
class SchemaRegistryClient:
    """Клиент для получения схем из registry."""
 
    def __init__(self, registry_url: str):
        self.registry_url = registry_url
        self.cache = {}
 
    @lru_cache(maxsize=128)
    def get_schema(self, subject: str, version: int | None = None) -> dict:
        """Получить схему из registry (с кэшированием)."""
        endpoint = f"/schemas/{subject}/versions/"
        endpoint += str(version) if version else "latest"
 
        response = requests.get(f"{self.registry_url}{endpoint}")
        response.raise_for_status()
 
        return response.json()
 
    def get_model(self, subject: str, version: int | None = None) -> type[BaseModel]:
        """Получить Pydantic модель из registry."""
        schema_data = self.get_schema(subject, version)
 
        # Динамическое создание модели из JSON Schema
        from pydantic import create_model
 
        fields = {}
        for field_name, field_info in schema_data['schema']['properties'].items():
            field_type = self._json_type_to_python(field_info['type'])
            fields[field_name] = (field_type, ...)
 
        return create_model(f"{subject.title()}V{schema_data['version']}", **fields)
 
    @staticmethod
    def _json_type_to_python(json_type: str):
        """Конвертация JSON типов в Python типы."""
        mapping = {
            'string': str,
            'integer': int,
            'number': float,
            'boolean': bool,
            'array': list,
            'object': dict,
        }
        return mapping.get(json_type, str)
 
# Использование в сервисе
registry = SchemaRegistryClient("http://schema-registry:8000")
 
# Получаем актуальную схему User
UserModel = registry.get_model("user")
 
# Валидируем данные
user_data = {"id": 1, "first_name": "Alice", "last_name": "Smith", "email": "alice@example.com"}
user = UserModel(**user_data)

# tests/test_contract_compatibility.py
import pytest
from my_company_schemas.user import User, UserV1
 
def test_v2_can_parse_v1_data():
    """V2 модель должна уметь парсить V1 данные."""
    v1_data = {
        "id": "123e4567-e89b-12d3-a456-426614174000",
        "name": "Alice Smith",
        "email": "alice@example.com",
        "created_at": "2025-01-01T00:00:00Z"
    }
 
    # V2 модель парсит v1 данные через миграцию
    user_v2 = User.model_validate(v1_data)
 
    assert user_v2.first_name == "Alice"
    assert user_v2.last_name == "Smith"
 
def test_v1_response_structure_unchanged():
    """V1 API response структура не должна меняться."""
    # Golden test: сохранённый пример реального ответа
    golden_response = {
        "id": "123e4567-e89b-12d3-a456-426614174000",
        "name": "Alice Smith",
        "email": "alice@example.com",
        "created_at": "2025-01-01T00:00:00Z"
    }
 
    # Проверяем, что UserV1 всё ещё парсит старый формат
    user = UserV1(**golden_response)
    serialized = user.model_dump()
 
    # Структура не изменилась
    assert set(serialized.keys()) == set(golden_response.keys())

# Плохо: неявное версионирование
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return user
 
# Хорошо: версия в URL
@app.get("/v2/users/{user_id}")
def get_user_v2(user_id: int):
    return user_v2
 
# Ещё лучше: версия через header с fallback
@app.get("/users/{user_id}")
def get_user(
    user_id: int,
    version: str = Header(default="v2", alias="Accept-Version")
):
    if version == "v1":
        return convert_to_v1(user)
    return user

## Breaking Changes in User Service v2.0
 
### Changed
 
- `name: str` → `first_name: str` + `last_name: str`
 
### Impact
 
Services that depend on User model:
 
- ✅ order-service: migrated to v2.0
- ⚠️ payment-service: still on v1.2, needs update
- ⚠️ notification-service: still on v1.0, needs update
 
### Migration Timeline
 
- 2025-01-15: v2.0 released, v1 deprecated (still supported)
- 2025-02-15: All services migrated to v2
- 2025-03-01: v1 support removed
 
### Migration Guide
 
See: docs/migrations/v1-to-v2.md

from prometheus_client import Counter
 
api_version_requests = Counter(
    'api_version_requests_total',
    'API requests by version',
    ['service', 'version', 'endpoint']
)
 
@app.middleware("http")
async def track_api_version(request: Request, call_next):
    version = request.headers.get('Accept-Version', 'unknown')
    service = request.headers.get('X-Service-Name', 'unknown')
 
    response = await call_next(request)
 
    api_version_requests.labels(
        service=service,
        version=version,
        endpoint=request.url.path
    ).inc()
 
    return response
 
# В Grafana видим:
# - Сколько сервисов ещё используют v1
# - Когда можно безопасно удалить deprecated версию

from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel, Field
from typing import Literal
 
app = FastAPI(
    title="User Management API",
    description="Production-ready API с полной Pydantic интеграцией",
    version="2.0.0"
)
 
class UserCreate(BaseModel):
    """Модель для создания пользователя.
 
    OpenAPI будет использовать этот docstring как описание схемы.
    """
    username: str = Field(
        min_length=3,
        max_length=20,
        pattern=r'^[a-zA-Z0-9_]+$',
        description="Уникальное имя пользователя",
        examples=["alice", "bob123"]
    )
    email: str = Field(
        description="Email адрес для уведомлений",
        examples=["alice@example.com"]
    )
    age: int = Field(
        ge=18,
        le=120,
        description="Возраст пользователя (18+)",
        examples=[25]
    )
 
class UserResponse(BaseModel):
    """Модель ответа с данными пользователя."""
    id: int = Field(description="Уникальный ID пользователя")
    username: str
    email: str
    age: int
    is_active: bool = Field(default=True, description="Статус активности")
 
    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "id": 1,
                    "username": "alice",
                    "email": "alice@example.com",
                    "age": 25,
                    "is_active": True
                }
            ]
        }
    )
 
@app.post(
    "/users",
    response_model=UserResponse,
    status_code=201,
    summary="Создать нового пользователя",
    description="Создаёт пользователя и возвращает его данные с присвоенным ID",
    response_description="Успешно созданный пользователь",
    tags=["Users"]
)
def create_user(user: UserCreate):
    # Pydantic автоматически валидирует и парсит request body
    # OpenAPI doc генерируется из UserCreate и UserResponse
    return UserResponse(
        id=123,
        username=user.username,
        email=user.email,
        age=user.age
    )

from typing import Union
from pydantic import BaseModel
 
class ErrorDetail(BaseModel):
    """Детали ошибки."""
    code: str = Field(description="Код ошибки")
    message: str = Field(description="Сообщение об ошибке")
    field: str | None = Field(default=None, description="Поле с ошибкой")
 
class ErrorResponse(BaseModel):
    """Стандартный формат ошибок API."""
    error: ErrorDetail
 
class SuccessResponse(BaseModel):
    """Успешный ответ."""
    data: UserResponse
    message: str = "Success"
 
@app.get(
    "/users/{user_id}",
    response_model=Union[SuccessResponse, ErrorResponse],
    responses={
        200: {
            "description": "Пользователь найден",
            "model": SuccessResponse
        },
        404: {
            "description": "Пользователь не найден",
            "model": ErrorResponse,
            "content": {
                "application/json": {
                    "example": {
                        "error": {
                            "code": "USER_NOT_FOUND",
                            "message": "User with ID 123 not found"
                        }
                    }
                }
            }
        }
    }
)
def get_user(user_id: int = Path(ge=1, description="ID пользователя")):
    user = db.query(User).get(user_id)
 
    if not user:
        return ErrorResponse(
            error=ErrorDetail(
                code="USER_NOT_FOUND",
                message=f"User with ID {user_id} not found"
            )
        )
 
    return SuccessResponse(
        data=UserResponse.model_validate(user)
    )

from enum import Enum
 
class SortOrder(str, Enum):
    """Порядок сортировки."""
    ASC = "asc"
    DESC = "desc"
 
class UserFilter(BaseModel):
    """Фильтры для списка пользователей."""
    search: str | None = Field(
        default=None,
        min_length=3,
        description="Поиск по username или email"
    )
    min_age: int | None = Field(default=None, ge=18, le=120)
    max_age: int | None = Field(default=None, ge=18, le=120)
    is_active: bool | None = Field(default=None)
    sort_by: Literal["username", "age", "created_at"] = "created_at"
    sort_order: SortOrder = SortOrder.DESC
    limit: int = Field(default=10, ge=1, le=100, description="Кол-во результатов")
    offset: int = Field(default=0, ge=0, description="Смещение для пагинации")
 
    @model_validator(mode='after')
    def validate_age_range(self):
        """Проверяем, что min_age <= max_age."""
        if self.min_age and self.max_age and self.min_age > self.max_age:
            raise ValueError('min_age must be less than or equal to max_age')
        return self
 
@app.get("/users", response_model=list[UserResponse])
def list_users(filters: UserFilter = Depends()):
    """Список пользователей с фильтрацией и сортировкой.
 
    Pydantic модель как dependency автоматически:
    - Парсит query parameters
    - Валидирует constraints
    - Генерирует OpenAPI doc
    """
    query = db.query(User)
 
    if filters.search:
        query = query.filter(
            or_(
                User.username.ilike(f'%{filters.search}%'),
                User.email.ilike(f'%{filters.search}%')
            )
        )
 
    if filters.min_age:
        query = query.filter(User.age >= filters.min_age)
 
    if filters.max_age:
        query = query.filter(User.age <= filters.max_age)
 
    if filters.is_active is not None:
        query = query.filter(User.is_active == filters.is_active)
 
    # Сортировка
    sort_column = getattr(User, filters.sort_by)
    if filters.sort_order == SortOrder.DESC:
        query = query.order_by(sort_column.desc())
    else:
        query = query.order_by(sort_column.asc())
 
    # Пагинация
    users = query.limit(filters.limit).offset(filters.offset).all()
 
    return [UserResponse.model_validate(u) for u in users]

from pydantic import BaseModel, Field, ConfigDict
from typing import Annotated
 
class UserProfile(BaseModel):
    """Профиль пользователя с кастомной JSON Schema."""
 
    model_config = ConfigDict(
        # Глобальные примеры для всей модели
        json_schema_extra={
            "examples": [
                {
                    "username": "alice",
                    "bio": "Software engineer",
                    "avatar_url": "https://example.com/avatars/alice.jpg",
                    "social_links": {
                        "github": "https://github.com/alice",
                        "twitter": "https://twitter.com/alice"
                    }
                }
            ]
        }
    )
 
    username: str = Field(
        min_length=3,
        json_schema_extra={
            "pattern": "^[a-zA-Z0-9_]+$",
            "title": "Username",
            "description": "Уникальное имя пользователя (только буквы, цифры, underscore)"
        }
    )
 
    bio: str | None = Field(
        default=None,
        max_length=500,
        json_schema_extra={
            "title": "Biography",
            "description": "Краткая биография пользователя"
        }
    )
 
    avatar_url: str | None = Field(
        default=None,
        json_schema_extra={
            "format": "uri",
            "title": "Avatar URL",
            "description": "URL аватара пользователя"
        }
    )
 
    social_links: dict[str, str] = Field(
        default_factory=dict,
        json_schema_extra={
            "title": "Social Links",
            "description": "Ссылки на социальные сети",
            "properties": {
                "github": {"type": "string", "format": "uri"},
                "twitter": {"type": "string", "format": "uri"},
                "linkedin": {"type": "string", "format": "uri"}
            }
        }
    )

from fastapi import APIRouter
 
# API v1
v1_router = APIRouter(prefix="/v1", tags=["API v1"])
 
class UserV1(BaseModel):
    """Старая версия модели."""
    name: str
    email: str
 
@v1_router.post("/users", response_model=UserV1, deprecated=True)
def create_user_v1(user: UserV1):
    """Создать пользователя (API v1 - deprecated)."""
    return user
 
# API v2
v2_router = APIRouter(prefix="/v2", tags=["API v2"])
 
class UserV2(BaseModel):
    """Новая версия модели."""
    first_name: str
    last_name: str
    email: str
 
@v2_router.post("/users", response_model=UserV2)
def create_user_v2(user: UserV2):
    """Создать пользователя (API v2 - current)."""
    return user
 
# Регистрация роутеров
app.include_router(v1_router)
app.include_router(v2_router)
 
# В OpenAPI будет 2 группы endpoints с тегами v1 (deprecated) и v2

from fastapi import HTTPException, status
 
class CreateUserSuccess(BaseModel):
    """Успешное создание пользователя."""
    id: int
    username: str
    message: str = "User created successfully"
 
class ValidationError(BaseModel):
    """Ошибка валидации."""
    detail: list[dict]
 
class ConflictError(BaseModel):
    """Конфликт (пользователь уже существует)."""
    error: str
    existing_user_id: int
 
@app.post(
    "/users",
    response_model=CreateUserSuccess,
    status_code=status.HTTP_201_CREATED,
    responses={
        201: {
            "description": "User created successfully",
            "model": CreateUserSuccess
        },
        422: {
            "description": "Validation error",
            "model": ValidationError
        },
        409: {
            "description": "User already exists",
            "model": ConflictError
        }
    }
)
def create_user_with_errors(user: UserCreate):
    """Создать пользователя с детальной документацией ошибок."""
 
    # Проверка уникальности
    existing = db.query(User).filter_by(username=user.username).first()
    if existing:
        raise HTTPException(
            status_code=409,
            detail=ConflictError(
                error="User already exists",
                existing_user_id=existing.id
            ).model_dump()
        )
 
    # Создание
    new_user = User(**user.model_dump())
    db.add(new_user)
    db.commit()
 
    return CreateUserSuccess(
        id=new_user.id,
        username=new_user.username
    )

from fastapi import FastAPI
 
app = FastAPI(
    title="Production API",
    description="API с полной Pydantic + OpenAPI интеграцией",
    version="2.0.0",
    openapi_tags=[
        {
            "name": "Users",
            "description": "Управление пользователями",
        },
        {
            "name": "Auth",
            "description": "Аутентификация и авторизация",
        },
        {
            "name": "Admin",
            "description": "Административные операции (требуют admin прав)",
        }
    ],
    openapi_url="/api/openapi.json",  # Кастомный URL для OpenAPI spec
    docs_url="/api/docs",              # Swagger UI
    redoc_url="/api/redoc"             # ReDoc
)
 
# Теперь все endpoints можно группировать по tags
@app.post("/users", tags=["Users"])
def create_user(user: UserCreate):
    pass
 
@app.post("/auth/login", tags=["Auth"])
def login(credentials: LoginCredentials):
    pass
 
@app.delete("/admin/users/{user_id}", tags=["Admin"])
def delete_user_admin(user_id: int):
    pass

# Экспорт OpenAPI spec для генерации TS типов
import json
from fastapi.openapi.utils import get_openapi
 
def export_openapi_schema():
    """Экспорт OpenAPI schema для генерации клиентского кода."""
    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        description=app.description,
        routes=app.routes,
    )
 
    with open("openapi.json", "w") as f:
        json.dump(openapi_schema, f, indent=2)
 
# Затем на клиенте:
# npx openapi-typescript openapi.json --output api.types.ts
 
# Результат - TypeScript типы:
# interface UserCreate {
#   username: string;
#   email: string;
#   age: number;
# }
#
# interface UserResponse {
#   id: number;
#   username: string;
#   email: string;
#   age: number;
#   is_active: boolean;
# }

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic import BaseModel, ConfigDict
from datetime import datetime
 
class Base(DeclarativeBase):
    pass
 
# SQLAlchemy ORM модель
class UserModel(Base):
    __tablename__ = "users"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(20), unique=True)
    email: Mapped[str] = mapped_column(String(255))
    password_hash: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
 
# Pydantic схема для чтения
class UserRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
 
    id: int
    username: str
    email: str
    is_active: bool
    created_at: datetime
    # password_hash намеренно пропущен (не возвращаем в API)
 
# Использование
@app.get("/users/{user_id}", response_model=UserRead)
def get_user(user_id: int):
    user = db.query(UserModel).get(user_id)
 
    # Автоматическая конвертация ORM → Pydantic
    return UserRead.model_validate(user)
 
    # Или короче (FastAPI сделает автоматически):
    # return user  # FastAPI вызовет UserRead.model_validate(user)

from sqlalchemy.orm import relationship
from pydantic import computed_field
 
class UserModel(Base):
    __tablename__ = "users"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]
 
    # Relationship
    posts: Mapped[list["PostModel"]] = relationship(back_populates="author")
 
class PostModel(Base):
    __tablename__ = "posts"
 
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
 
    author: Mapped["UserModel"] = relationship(back_populates="posts")
 
class UserWithStats(BaseModel):
    model_config = ConfigDict(from_attributes=True)
 
    id: int
    username: str
 
    @computed_field
    @property
    def posts_count(self) -> int:
        """Вычисляемое поле: количество постов."""
        # Доступ к relationship из ORM модели
        return len(self.posts) if hasattr(self, 'posts') else 0
 
# Использование
@app.get("/users/{user_id}/stats", response_model=UserWithStats)
def get_user_stats(user_id: int):
    # Eager loading relationship для computed field
    user = db.query(UserModel).options(
        joinedload(UserModel.posts)
    ).get(user_id)
 
    return UserWithStats.model_validate(user)
    # { "id": 1, "username": "alice", "posts_count": 15 }