Tutorial — construindo a feature Users

Este tutorial passa pela conexão da feature Users usando todas as convenções do SDK. Ao final você terá:

• Um UserModel SQLAlchemy com colunas de auditoria + soft-delete
• Schemas Pydantic para create / update / response / filter
• Um repository, service e controller herdando das bases do SDK
• Routers com controllers injetados via Depends
• Subclasses de exceção de domínio serializadas pelo exception handler do SDK
• Um GET /users paginado e um POST /users de signup (com hashing de senha via PasswordUtils)

Para os impacientes

Se você só quer copiar o layout, gere-o: tempest new my-service. A CLI entrega o mesmo esqueleto que este tutorial percorre.

Já tem fluxo de auth pronto

Este tutorial mostra como construir signup/login com BaseRepository + BaseService + BaseController — é a base para qualquer feature. Para o fluxo de auth completo (signup + activation por email + login com JWT + reset de senha), o SDK fornece UserAuthService + make_auth_router desde v0.31.0; pule para a receita Auth flow » quando quiser usar o atalho em vez de implementar manualmente.

Acompanhando

Snippets de arquivo inteiro — os que começam com um comentário de caminho de arquivo isolado (ex.: # src/db/models/user.py) — são standalone: cole cada um no caminho indicado. Blocos marcados com (continued) ou com um comentário # ... above ... anexam ao arquivo mostrado logo acima, não são um arquivo novo. A árvore completa do projeto é o layout obrigatório de projeto da Arquitetura →.

Vamos construir uma feature Users completa do zero, de ponta a ponta. Todo arquivo abaixo é algo que você escreve no seu projeto; os primitivos do SDK são importados.

1. Layout do projeto

O layout canônico que todo serviço Python distribuído contra este SDK deve adotar — main.py é um one-liner, src/server.py expõe tanto run() quanto o app importável (ou o re-exporta de src/api/app.py), api/dependencies/ é sempre um pacote (auth + provedores factory), controllers/ é obrigatório mesmo quando é só um pass-through fino, e repositories/ vive sob db/.

my-service/
├── main.py                       # one-liner: from src.server import run; run()
└── src/
    ├── __init__.py               # re-exporta `run` de src.server
    ├── server.py                 # uvicorn.run(...) programático + `app` no nível do módulo
    ├── core/
    │   ├── __init__.py
    │   ├── settings.py           # Settings(mixins..., BaseAppSettings)
    │   └── exceptions.py         # exceções de domínio (UserNotFoundError, ...)
    ├── db/
    │   ├── __init__.py           # re-exporta BaseModel + todo modelo
    │   ├── models/
    │   │   ├── __init__.py
    │   │   └── user.py           # UserModel(BaseModel)
    │   └── repositories/
    │       ├── __init__.py
    │       └── user.py           # UserRepository(BaseRepository[UserModel])
    ├── schemas/
    │   ├── __init__.py
    │   └── user.py               # UserCreate/Update/Response/Filter
    ├── services/
    │   ├── __init__.py
    │   └── user.py               # UserService — lógica de negócio
    ├── controllers/
    │   ├── __init__.py
    │   └── user.py               # UserController — orquestração (pass-through fino OK)
    └── api/
        ├── __init__.py
        ├── app.py                # create_app() — middleware, CORS, exception handlers, routers
        ├── routers/
        │   ├── __init__.py
        │   └── users.py
        └── dependencies/         # SEMPRE um pacote, nunca um módulo plano
            ├── __init__.py
            ├── auth.py           # dependências X-Token / current_user / require_role
            ├── resources.py      # singletons de infra (db/storage/mail) + get_db/get_session
            └── controllers.py    # factories get_<X>_controller / get_<X>_service

Cada __init__.py re-exporta todo símbolo público do seu diretório para que os consumidores sempre façam from src.schemas import UserCreateSchema (não from src.schemas.user import UserCreateSchema). Isso mantém os refactors indolores — mova arquivos sem quebrar imports.

Se o seu serviço ainda não tem controllers/services/repositories, ainda assim distribua pacotes vazios com os nomes certos — a uniformidade importa mais do que pular um diretório. Descarte db/, utils/, queue/ ou tasks/ só quando o serviço genuinamente não precisa de persistência/utilitários/mensageria.

2. Settings, server, factory do app & entrypoint

Quatro arquivos mapeiam em quatro responsabilidades:

Arquivo	Responsabilidade
src/core/settings.py	Settings(...mixins, BaseAppSettings) — uma fonte única de verdade para env vars.
src/api/dependencies/resources.py	singletons de infra (db = AsyncDatabaseManager(**settings.database_kwargs()), e — opt-in — storage/mail) + provedores get_db / get_session. Dono único do ciclo de vida dos recursos.
src/api/app.py	factory create_app() magra — middleware + CORS + exception handlers + includes de router + instância app no nível do módulo. Importa os recursos de dependencies, não os constrói.
src/server.py	run() invocando uvicorn.run("src.api.app:app", ...) programaticamente, mais re-exporta app para que runners externos (gunicorn, CLI do uvicorn) possam importá-lo.
main.py	Entry point do processo — uma única linha sob if __name__ == "__main__": chamando run().

# src/core/settings.py
from tempest_fastapi_sdk import BaseAppSettings, DatabaseSettings, ServerSettings


class Settings(ServerSettings, DatabaseSettings, BaseAppSettings):
    """All environment-driven configuration lives here.

    BaseAppSettings ships `env_file=.env`, `extra=ignore`,
    `case_sensitive=True`, `frozen=True` and `str_strip_whitespace=True`.
    ServerSettings adds SERVER_HOST/PORT/RELOAD, DatabaseSettings adds
    DATABASE_URL/ECHO/POOL_*.
    """

    JWT_SECRET: str
    JWT_ALGORITHM: str = "HS256"
    JWT_TTL_HOURS: int = 1

    SMTP_HOST: str = "localhost"
    SMTP_PORT: int = 587
    SMTP_USERNAME: str | None = None
    SMTP_PASSWORD: str | None = None
    SMTP_FROM_ADDR: str = "noreply@example.com"

    UPLOAD_DIR: str = "./var/uploads"


settings = Settings()

BaseAppSettings é sempre a última base

Todo mixin de settings do SDK (ServerSettings, DatabaseSettings, RedisSettings, …) já herda BaseAppSettings — é assim que cada um carrega o model_config canônico (env_file=".env", extra="ignore", case_sensitive=True) independente da ordem.

Consequência: BaseAppSettings tem que vir por último na lista de bases. Python proíbe uma base preceder a própria subclasse (linearização C3), então listar

# docs-guard: skip — este é o erro descrito no aviso
class Settings(BaseAppSettings, DatabaseSettings):  # ❌ não importa
    ...

falha em tempo de import com:

TypeError: Settings: BaseAppSettings must be the LAST base —
DatabaseSettings already subclasses it, so listing BaseAppSettings
before it is an invalid method resolution order (MRO). Move
BaseAppSettings to the end of the base list:
class Settings(DatabaseSettings, BaseAppSettings).

Antes da v0.159.1 a mensagem era o TypeError cru do pydantic (Cannot create a consistent method resolution order (MRO) for bases BaseAppSettings, DatabaseSettings), acompanhado no mypy de um [metaclass] enganoso. A causa é a mesma nas duas versões: mova BaseAppSettings para o fim.

Os recursos de infra (o banco agora; storage / mail quando entrarem) vivem em src/api/dependencies/resources.py, construídos uma vez e acessados por provedores get_*. O app.py só os importa — assim ele não engorda com construção de recursos:

# src/api/dependencies/resources.py
from tempest_fastapi_sdk import AsyncDatabaseManager

from src.core.settings import settings

db = AsyncDatabaseManager(**settings.database_kwargs())
get_session = db.session_dependency   # Depends(get_session) -> AsyncSession


def get_db() -> AsyncDatabaseManager:
    """Return the process-wide database manager."""
    return db

# src/api/app.py
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from tempest_fastapi_sdk import (
    RequestIDMiddleware,
    make_health_router,
    register_exception_handlers,
)

from src.api.dependencies.resources import db
from src.api.routers import users


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    """Connect on startup, dispose on shutdown."""
    await db.connect()
    try:
        yield
    finally:
        await db.disconnect()


def create_app() -> FastAPI:
    """Build and configure the FastAPI app."""
    app = FastAPI(title="my-service", version="0.1.0", lifespan=lifespan)

    app.add_middleware(RequestIDMiddleware)
    register_exception_handlers(app)

    # Meta endpoints sit at the root prefix.
    app.include_router(make_health_router(db=db, version="0.1.0"))

    # Business endpoints sit under /api/<domain>.
    app.include_router(users.router, prefix="/api")
    return app


app = create_app()

# src/server.py
from tempest_fastapi_sdk import run_server

from src.api.app import app  # noqa: F401 — re-exported for external runners
from src.core.settings import settings


def run() -> None:
    """Start the API server programmatically."""
    run_server("src.api.app:app", settings=settings)


__all__: list[str] = ["app", "run"]

run_server lê SERVER_HOST / SERVER_PORT / SERVER_RELOAD de settings (caindo para 127.0.0.1 / 8000 / False) e encaminha quaisquer kwargs extras (workers=, log_config=, ssl_*=) literalmente para uvicorn.run. Veja a receita de ponto de entrada programático do servidor.

# src/__init__.py
from src.server import run

__all__: list[str] = ["run"]

# main.py
from src.server import run

if __name__ == "__main__":
    run()

Defaults de bind: 127.0.0.1 para serviços internos (o default ServerSettings.SERVER_HOST do SDK), 0.0.0.0 só quando o serviço é consumido por uma origem separada (ex.: um dev server de frontend). Nunca inicie o uvicorn via subprocess.run(["uvicorn", ...]) — sempre passe por run_server (ou uvicorn.run("src.api.app:app", ...) diretamente) para que reload, tratamento de sinais e shutdown gracioso se comportem corretamente.

3. Modelo ORM

# src/db/models/user.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel


class UserModel(BaseModel):
    """One row per registered user.

    Inherits from BaseModel, so it automatically gets:
    - id (UUID v4, cross-DB portable via sqlalchemy.Uuid)
    - is_active (bool, soft-delete flag)
    - created_at, updated_at (timezone-aware TIMESTAMP, set by Python AND
      the DB so the instance attribute is populated right after flush)
    - __tablename__ = "user" (auto: class name without "Model" suffix,
      snake-cased; override by assigning __tablename__ explicitly)
    - __eq__/__hash__ by (type, id) so the same row across sessions
      compares equal
    - to_dict(exclude, include, remove_none) and
      update_from_dict(data, allowed_fields) helpers
    """

    name: Mapped[str] = mapped_column(String(64), nullable=False)
    email: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(128), nullable=False)

Re-exporte:

# src/db/models/__init__.py
from src.db.models.user import UserModel

__all__: list[str] = ["UserModel"]

# src/db/__init__.py
from src.db.models import UserModel
from tempest_fastapi_sdk import BaseModel

__all__: list[str] = ["BaseModel", "UserModel"]

Dica: Sempre importe os modelos em src/db/__init__.py. O SQLAlchemy precisa "ver" todo modelo antes de BaseModel.metadata ficar completa, para que o autogenerate do Alembic e o create_tables() funcionem corretamente.

4. Schemas

Instalação

Os schemas abaixo usam EmailStr, que precisa do validador email-validator — traga-o com o extra [email] (uv add "tempest-fastapi-sdk[email]") ou via pydantic[email]. Mais adiante o service usa PasswordUtils (hashing bcrypt), que depende do extra [auth] — uv add "tempest-fastapi-sdk[auth]".

O padrão de nomenclatura recomendado: um schema *Create, *Update, *Response e *Filter por recurso.

# src/schemas/user.py
from pydantic import EmailStr, Field

from tempest_fastapi_sdk import (
    BasePaginationFilterSchema,
    BaseResponseSchema,
    BaseSchema,
)


class UserCreateSchema(BaseSchema):
    """Payload for POST /users."""

    name: str = Field(min_length=1, max_length=64)
    email: EmailStr
    password: str = Field(min_length=8, max_length=128)


class UserUpdateSchema(BaseSchema):
    """Partial payload for PATCH /users/{id}. Every field optional."""

    name: str | None = Field(default=None, min_length=1, max_length=64)
    email: EmailStr | None = None


class UserResponseSchema(BaseResponseSchema):
    """Outbound representation.

    Inherits id/is_active/created_at/updated_at from BaseResponseSchema
    (timestamps already normalized to UTC by the field validator).
    """

    name: str
    email: EmailStr


class UserFilterSchema(BasePaginationFilterSchema):
    """Query-string filters for GET /users.

    Inherits page/page_size/order_by/ascending/is_active from
    BasePaginationFilterSchema. Add domain-level filters below.
    """

    name: str | None = None              # ILIKE %name% search
    email: EmailStr | None = None        # exact-match filter

# src/schemas/__init__.py
from src.schemas.user import (
    UserCreateSchema,
    UserFilterSchema,
    UserResponseSchema,
    UserUpdateSchema,
)

__all__: list[str] = [
    "UserCreateSchema",
    "UserFilterSchema",
    "UserResponseSchema",
    "UserUpdateSchema",
]

5. Exceções de domínio

O SDK entrega NotFoundException, ConflictException, etc. genéricos. Subclasse-os por domínio para que o matching isinstance / except DomainError fique explícito. message / code / status_code a nível de classe são defaults aos quais o construtor recorre — você também pode sobrescrever qualquer um deles no ponto do raise sem subclassear:

# src/core/exceptions.py
from tempest_fastapi_sdk import ConflictException, NotFoundException


class UserNotFoundError(NotFoundException):
    """Subclass kept only for ``except UserNotFoundError`` matching."""

    message: str = "Usuário não encontrado"
    code: str = "USER_NOT_FOUND"


class UserEmailAlreadyTakenError(ConflictException):
    message: str = "Já existe um usuário com esse e-mail"
    code: str = "USER_EMAIL_TAKEN"

Para códigos pontuais você não precisa de uma subclasse — passe-os ao construtor:

from uuid import UUID

from tempest_fastapi_sdk import NotFoundException

order_id = UUID("6f1c3d84-2a55-4d0b-9d7e-0c1a2b3c4d5e")


raise NotFoundException(
    "Pedido não encontrado",
    code="ORDER_NOT_FOUND",
    details={"order_id": str(order_id)},
)

O exception handler do SDK (register_exception_handlers) os serializa para:

{
    "detail": "Usuário não encontrado",
    "code": "USER_NOT_FOUND",
    "details": {}
}

O frontend ramifica no code, não na mensagem (que pode estar traduzida).

6. Repository

Para CRUD simples você não precisa de uma subclasse nenhuma — instancie BaseRepository diretamente e vincule o modelo pelo construtor:

# anywhere a session is in scope

import asyncio

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from tempest_fastapi_sdk import BaseRepository

from src.db.models import UserModel


# Num serviço, a sessão real vem de `db.get_session_context()`; aqui, do SQLite.
session = AsyncSession(create_async_engine("sqlite+aiosqlite:///:memory:"))

repository = BaseRepository(session, model=UserModel)


async def main() -> None:
    """Run this example."""
    await repository.add(
        UserModel(
            email="ana@example.com",
            name="Ana",
            password_hash="<bcrypt-hash>",
        )
    )


asyncio.run(main())

Subclasse quando quiser embutir mensagens específicas de domínio, trocar a exceção de not-found, sobrescrever os métodos de mapeamento ou adicionar queries custom. A assinatura do construtor (não os atributos de classe) é o contrato:

# src/db/repositories/user.py
from sqlalchemy.ext.asyncio import AsyncSession

from tempest_fastapi_sdk import BaseRepository

from src.core.exceptions import UserNotFoundError
from src.db.models import UserModel
from src.schemas import UserResponseSchema


class UserRepository(BaseRepository[UserModel]):
    """Data-access layer for users."""

    def __init__(self, session: AsyncSession) -> None:
        super().__init__(
            session,
            model=UserModel,
            not_found_exception=UserNotFoundError,
            not_found_message="Usuário não encontrado",
            create_conflict_message="Já existe um usuário com esse e-mail",
            update_conflict_message="Conflito ao atualizar usuário",
        )

    def map_to_schema(self, instance: UserModel) -> UserResponseSchema:
        return UserResponseSchema.model_validate(instance)

    def map_to_response(self, instance: UserModel) -> UserResponseSchema:
        return self.map_to_schema(instance)

O repo base te dá 20+ métodos de graça — veja a tabela de referência abaixo. Adicione queries custom em cima do mesmo UserRepository:

# src/db/repositories/user.py  (continued)

from tempest_fastapi_sdk import BaseRepository

from src.db.models import UserModel


class UserRepository(BaseRepository[UserModel]):
    # ... __init__ and mappers above ...

    # ──────── custom queries on top of the inherited bulk + read methods ────────

    async def get_by_email(self, email: str) -> UserModel:
        """Look up a user by email. Raises ``UserNotFoundError`` on miss."""
        return await self.get({"email": email})

O bloco destacado (sob o comentário divisor) é o que você tipicamente adiciona por projeto — tudo acima dele é o boilerplate de que a classe base já cuida.

7. Service

O service é onde as regras de negócio vivem. Ele chama um ou mais repositories e nunca toca em tipos de HTTP ou SQLAlchemy diretamente.

Herde de BaseService[RepositoryT, ResponseT]. Fazer isso te dá get_by_id, get_or_none, list, paginate, count, exists e delete de graça — cada um já conectado a repository.map_to_response (sync ou async). Sobrescreva só os métodos que precisam de lógica de domínio; adicione novos para casos de uso que a base não cobre (signup, reset de senha, etc.):

# src/services/user.py
from uuid import UUID

from sqlalchemy.ext.asyncio import AsyncSession

from tempest_fastapi_sdk import BaseService, PasswordUtils

from src.core.exceptions import UserEmailAlreadyTakenError
from src.db.repositories import UserRepository
from src.schemas import UserCreateSchema, UserResponseSchema, UserUpdateSchema


class UserService(BaseService[UserRepository, UserResponseSchema]):
    """Business logic for the user domain.

    Inherits the canonical read-path methods (``get_by_id`` / ``list`` /
    ``paginate`` / ``count`` / ``exists`` / ``delete``) from
    :class:`BaseService` and adds the write-path methods that need
    domain rules (uniqueness check, password hashing, mass-assignment
    guard).
    """

    def __init__(
        self,
        repository: UserRepository,
        *,
        passwords: PasswordUtils,
    ) -> None:
        """Initialize the service.

        Args:
            repository (UserRepository): User-domain repository.
            passwords (PasswordUtils): Shared bcrypt helper.
        """
        super().__init__(repository)
        self.passwords: PasswordUtils = passwords

    # ──────── overrides: domain rules live here ────────

    async def signup(self, data: UserCreateSchema) -> UserResponseSchema:
        """Create a new user, enforcing email uniqueness + hashing the password."""
        if await self.repository.exists({"email": data.email}):
            raise UserEmailAlreadyTakenError()
        instance = self.repository.map_to_model(
            {
                **data.to_dict(exclude=["password"]),
                "password_hash": self.passwords.hash(data.password),
            },
        )
        instance = await self.repository.add(instance)
        return self.repository.map_to_response(instance)

    async def update(
        self,
        user_id: UUID,
        data: UserUpdateSchema,
    ) -> UserResponseSchema:
        """Apply a partial update, whitelisting the columns that may change."""
        instance = await self.repository.get_by_id(user_id)
        instance.update_from_dict(
            data.to_dict(),
            allowed_fields={"name", "email"},   # prevents mass-assignment
        )
        instance = await self.repository.update(instance)
        return self.repository.map_to_response(instance)

    async def soft_delete(self, user_id: UUID) -> None:
        """Flip ``is_active=False`` instead of hard-deleting."""
        await self.repository.soft_delete(user_id)

Os métodos que você não escreve — get_by_id(user_id), get_or_none(filters), list(filters=None, order_by=None, ascending=True), paginate(filters=None, order_by=None, page=1, page_size=20, ascending=True), count(filters), exists(filters), delete(user_id) — já existem na base, já aguardam um map_to_response async, e já retornam o UserResponseSchema tipado declarado no parâmetro genérico.

Quando o caso de uso precisa de um pipeline custom (joins, projeções, fan-out transacional), sobrescreva o método herdado. A assinatura continua a mesma, então o controller não percebe:

from typing import Any

from tempest_fastapi_sdk import BaseService

from src.db.repositories import UserRepository
from src.schemas import UserResponseSchema


class UserService(BaseService[UserRepository, UserResponseSchema]):
    # ... __init__ and overrides above ...

    async def list(  # override the inherited pass-through
        self,
        filters: dict[str, Any] | None = None,
        order_by: Any | None = None,
        ascending: bool = True,
    ) -> list[UserResponseSchema]:
        """List active users only — domain rule baked into the base method."""
        merged: dict[str, Any] = {**(filters or {}), "is_active": True}
        return await super().list(filters=merged, order_by=order_by, ascending=ascending)

8. Controller

Mesmo quando não há orquestração a fazer, controllers/ existe como um pass-through fino para que o grafo de imports fique uniforme entre os serviços. No dia em que um caso de uso precisar coordenar dois services (ou fazer fan-out para uma fila), o controller já está lá.

Herde de BaseController[ServiceT, ResponseT]. A base encaminha get_by_id, list, paginate, count e delete para o service por você — você só declara métodos que adicionam coordenação entre services ou que não existem no service (casos de uso custom como signup):

# src/controllers/user.py
from uuid import UUID

from tempest_fastapi_sdk import BaseController

from src.schemas import UserCreateSchema, UserResponseSchema, UserUpdateSchema
from src.services.user import UserService


class UserController(BaseController[UserService, UserResponseSchema]):
    """Orchestrate user use cases.

    Today every method is a thin pass-through to ``UserService``. As
    soon as a use case needs to coordinate more than one service —
    e.g. signup also sends a welcome email and enqueues a CRM sync —
    the orchestration lives here, not in the router and not in the
    service.
    """

    # ──────── new methods for use cases the base doesn't cover ────────

    async def signup(self, data: UserCreateSchema) -> UserResponseSchema:
        """Create a user and (eventually) trigger downstream side effects."""
        return await self.service.signup(data)

    async def update(
        self,
        user_id: UUID,
        data: UserUpdateSchema,
    ) -> UserResponseSchema:
        """Domain-specific partial update — distinct from the base ``delete``."""
        return await self.service.update(user_id, data)

    async def soft_delete(self, user_id: UUID) -> None:
        """Soft-delete instead of the inherited hard ``delete``."""
        await self.service.soft_delete(user_id)

get_by_id / list / paginate / count não são redeclarados — BaseController já os expõe. Quando o dia da coordenação entre services chegar, sobrescreva o pass-through no lugar:

from tempest_fastapi_sdk import BaseController

from src.schemas import UserCreateSchema, UserResponseSchema
from src.services import UserService


class UserController(BaseController[UserService, UserResponseSchema]):
    # ... methods above ...

    async def signup(self, data: UserCreateSchema) -> UserResponseSchema:
        """Create the user, send a welcome email, enqueue the CRM sync."""
        user = await self.service.signup(data)
        await self.emails.send_welcome(user)            # second dependency
        await self.tasks.enqueue("crm.user.created", {"id": str(user.id)})
        return user

A assinatura do router nunca muda — só o corpo do controller cresce.

9. Provedores de dependência

api/dependencies/ é sempre um pacote. auth.py hospeda as dependências de segredo compartilhado / usuário atual; resources.py hospeda os singletons de infra (db, e — opt-in — storage/mail) com os provedores get_db / get_session; controllers.py (ou services.py quando ainda não há camada de controller) hospeda os provedores factory dos quais os routers dependem. Nunca construa controllers, services ou recursos de infra inline dentro do arquivo do router (nem do app.py).

# src/api/dependencies/controllers.py
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from tempest_fastapi_sdk import PasswordUtils

from src.api.dependencies.resources import get_session
from src.controllers.user import UserController
from src.db.repositories import UserRepository
from src.services.user import UserService


# Stateless utilities — instantiate once per process.
_passwords: PasswordUtils = PasswordUtils()


def get_user_controller(
    session: AsyncSession = Depends(get_session),
) -> UserController:
    """Wire repository → service → controller for a single request."""
    repository = UserRepository(session)
    service = UserService(repository=repository, passwords=_passwords)
    return UserController(service=service)

# src/api/dependencies/__init__.py
from src.api.dependencies.controllers import get_user_controller

__all__: list[str] = ["get_user_controller"]

10. Router

Routers recebem controllers via Depends do FastAPI — sem construção inline, sem lógica de negócio, sem chamadas de DB. Endpoints de negócio ficam sob /api/<domínio> (o prefixo é adicionado no ponto do include em src/api/app.py); endpoints meta (/health, /tool-spec) ficam no prefixo raiz.

# src/api/routers/users.py
from uuid import UUID

from fastapi import APIRouter, Depends, status

from tempest_fastapi_sdk import BasePaginationSchema

from src.api.dependencies import get_user_controller
from src.controllers.user import UserController
from src.schemas import (
    UserCreateSchema,
    UserFilterSchema,
    UserResponseSchema,
    UserUpdateSchema,
)


router = APIRouter(prefix="/users", tags=["users"])


@router.post(
    "",
    response_model=UserResponseSchema,
    status_code=status.HTTP_201_CREATED,
)
async def create_user(
    data: UserCreateSchema,
    controller: UserController = Depends(get_user_controller),
) -> UserResponseSchema:
    return await controller.signup(data)


@router.get("/{user_id}", response_model=UserResponseSchema)
async def get_user(
    user_id: UUID,
    controller: UserController = Depends(get_user_controller),
) -> UserResponseSchema:
    return await controller.get_by_id(user_id)


@router.patch("/{user_id}", response_model=UserResponseSchema)
async def update_user(
    user_id: UUID,
    data: UserUpdateSchema,
    controller: UserController = Depends(get_user_controller),
) -> UserResponseSchema:
    return await controller.update(user_id, data)


@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: UUID,
    controller: UserController = Depends(get_user_controller),
) -> None:
    await controller.soft_delete(user_id)


@router.get("", response_model=BasePaginationSchema[UserResponseSchema])
async def list_users(
    filters: UserFilterSchema = Depends(),
    controller: UserController = Depends(get_user_controller),
) -> BasePaginationSchema[UserResponseSchema]:
    result = await controller.paginate(
        filters=filters.get_conditions(),
        order_by=filters.order_by,
        page=filters.page,
        page_size=filters.page_size,
        ascending=filters.ascending,
    )
    return BasePaginationSchema[UserResponseSchema](**result)

11. Paginação

O contrato de paginação é imposto de ponta a ponta pelos primitivos do SDK:

• UserFilterSchema(BasePaginationFilterSchema) parseia ?page=&page_size=&order_by=&ascending=&is_active=&name= da query string e expõe .get_conditions() retornando só os filtros de domínio (sem as chaves de paginação).
• UserRepository.paginate(...) roda a query com o dict de filtro + ordenação + offset/limit + contagem, retornando o dict {items, total, page, page_size, pages} que você embrulha em BasePaginationSchema[UserResponseSchema].
• BasePaginationSchema[UserResponseSchema] embrulha o resultado para que o OpenAPI documente o formato da resposta corretamente.

GET /api/users?page=2&page_size=20&order_by=name&ascending=true&is_active=true&name=ana

Retorna:

{
    "items": [
        {"id": "...", "name": "Ana ...", "email": "...", ...},
        ...
    ],
    "total": 142,
    "page": 2,
    "page_size": 20,
    "pages": 8
}

---

Revisão

Você conectou a feature Users de ponta a ponta, uma camada por vez, cada uma apoiada num primitivo do SDK:

• Modelo — UserModel(BaseModel) ganha id / is_active / created_at / updated_at + helpers de graça.
• Schemas — UserCreateSchema / UserUpdateSchema / UserResponseSchema / UserFilterSchema sobre BaseSchema / BaseResponseSchema / BasePaginationFilterSchema.
• Exceções — subclasses de domínio (UserNotFoundError, UserEmailAlreadyTakenError) serializadas pelo register_exception_handlers.
• Repository — UserRepository(BaseRepository[UserModel]) — 20+ métodos herdados + queries custom.
• Service — UserService(BaseService[...]) — regras de negócio (unicidade de e-mail, hash de senha) sobre o read-path herdado.
• Controller — UserController(BaseController[...]) — orquestração (pass-through fino hoje, coordenação amanhã).
• Router — endpoints com UserController injetado via Depends, montados sob /api/users.
• Paginação — BasePaginationFilterSchema → paginate(...) → BasePaginationSchema[UserResponseSchema] de ponta a ponta.

Próximos passos

• Fluxo de auth » — troque o signup manual pelo UserAuthService + make_auth_router (signup + ativação por e-mail + login JWT + reset de senha).
• Banco de dados » — migrations com Alembic, operações em lote e os mixins de auditoria/soft-delete do BaseRepository.
• Testes » — teste a feature com pytest + SQLite in-memory.
• Ponto de entrada do servidor » — middleware, CORS, health, tool-spec e run_server.
• Segurança » — endureça o serviço (rate limit, CSRF, idempotência, limites de corpo).