backend-python-fastapi

# Backend: Python + FastAPI + Hexagonal > Exemplos de Value Objects Python na seção abaixo. ## Estrutura de pastas > **SOLID e Clean Code são não-negociáveis.** DIP já está expresso nas portas e adaptadores (hexagonal). Mantenha SRP: cada classe tem uma responsabilidade. Nomes de métodos devem ser autoexplicativos — prefira verbos longos a abreviações. ``` app/ domain/ entity/ # dataclasses com comportamento de negócio valueobject/ # email.py, cpf.py, status.py (Enum com transições) event/ # Domain Events usecase/ command/ <entidade>/ # ex: processo/, usuario/ query/ <entidade>/ # ex: processo/, usuario/ port/ # Protocol (interfaces) — ProcessoRepository, Cache, EventPublisher adapter/ http/ # APIRouter do FastAPI (entrada) sqlalchemy/ # implementação dos ports de persistência (saída) redis/ # implementação do Cache main.py # bootstrap: cria FastAPI + registra routers ``` Regra inegociável: `domain` e `usecase` NUNCA importam de `adapter`. ## Ports como typing.Protocol Em Python, `Protocol` substitui ABC/interfaces formais — duck typing estrutural: ```python # port/processo_repository.py from typing import Protocol from uuid import UUID from app.domain.entity.processo import Processo class ProcessoRepository(Protocol): async def salvar(self, processo: Processo) -> None: ... async def buscar_por_id(self, id: UUID) -> Processo | None: ... async def listar(self, page: int, page_size: int, sort: str, order: str) -> tuple[list[Processo], int]: ... ``` ## Value Objects com Pydantic (imutável, valida no construtor) ```python # domain/valueobject/email.py from pydantic import BaseModel, field_validator class Email(BaseModel): valor: str model_config = {"frozen": True} # imutável — Pydantic v2 @field_validator("valor") @classmethod def validar(cls, v: str) -> str: v = v.strip().lower() if "@" not in v or "." not in v.split("@")[-1]: raise ValueError(f"Email inválido: {v!r}") return v def __str__(self) -> str: return self.valor ``` Status com transições válidas: ```python # domain/valueobject/status_processo.py from enum import Enum _TRANSICOES = { "aberto": ["em_analise", "cancelado"], "em_analise": ["concluido", "cancelado"], "concluido": [], "cancelado": [], } class StatusProcesso(str, Enum): ABERTO = "aberto" EM_ANALISE = "em_analise" CONCLUIDO = "concluido" CANCELADO = "cancelado" def transicionar_para(self, proximo: "StatusProcesso") -> "StatusProcesso": if proximo.value not in _TRANSICOES[self.value]: raise ValueError(f"Transição inválida: {self} → {proximo}") return proximo ``` ## Injeção de dependência com FastAPI Depends ```python # adapter/http/processo_router.py from fastapi import APIRouter, Depends from app.port.processo_repository import ProcessoRepository from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine from app.adapter.sqlalchemy.processo_repository import ProcessoSqlRepository router = APIRouter(prefix="/api/v1/processos") async def get_conn(engine: AsyncEngine = Depends(get_engine)): async with engine.connect() as conn: yield conn def get_repository(conn: AsyncConnection = Depends(get_conn)) -> ProcessoRepository: return ProcessoSqlRepository(conn) # substituível por mock em testes @router.post("/", status_code=201) async def criar( body: CriarProcessoRequest, repo: ProcessoRepository = Depends(get_repository), ): use_case = CriarProcessoUseCase(repo) return await use_case.executar(body.to_command()) ``` ## Persistência com SQLAlchemy Core (sem ORM) Dependências: `sqlalchemy[asyncio]` e `asyncpg`. Usar SQLAlchemy **Core** (SQL Expression Language) — nunca o ORM declarativo (`declarative_base`, `Session`, `query()`, `relationship()`, `backref`). ```python # infrastructure/database.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncConnection from sqlalchemy import Table, Column, String, DateTime, MetaData metadata = MetaData() processos = Table( "processos", metadata, Column("id", String, primary_key=True), Column("numero", String, nullable=False), Column("status", String, nullable=False), Column("criado_em", DateTime(timezone=True), nullable=False), ) ``` ```python # adapter/sqlalchemy/processo_repository.py from sqlalchemy.dialects.postgresql import insert as pg_insert class ProcessoSqlRepository(ProcessoRepository): def __init__(self, conn: AsyncConnection): self._conn = conn async def find_by_id(self, id: str) -> Processo | None: result = await self._conn.execute( processos.select().where(processos.c.id == id) ) row = result.fetchone() return ProcessoMapper.to_domain(row) if row else None async def save(self, processo: Processo) -> None: # on_conflict_do_update só existe no insert do dialeto PostgreSQL await self._conn.execute( pg_insert(processos) .values(ProcessoMapper.to_persistence(processo)) .on_conflict_do_update( index_elements=["id"], set_=ProcessoMapper.to_persistence(processo), ) ) ``` Listagem com paginação e ordenação: ```python # adapter/sqlalchemy/processo_repository.py COLUNAS_ORDENAVEIS = {"criado_em", "descricao", "status"} async def listar(self, page, page_size, sort, order): if sort and sort not in COLUNAS_ORDENAVEIS: raise HTTPException(status_code=400, detail=f"Ordenação inválida: {sort}") coluna = sort or "criado_em" ordem = "DESC" if order == "desc" else "ASC" resultado = await self._conn.execute( text(f"SELECT ... FROM processo WHERE excluido_em IS NULL " f"ORDER BY {coluna} {ordem} LIMIT :limit OFFSET :offset"), {"limit": page_size, "offset": (page - 1) * page_size}, ) ``` Allowlist obrigatória — nunca interpolar `sort` diretamente na query. ## OpenAPI / Swagger (FastAPI gera automaticamente — nenhuma lib extra) FastAPI gera `openapi.json` automaticamente a partir das type hints e modelos Pydantic. Configure título, versão e descrição uma única vez: ```python # main.py from fastapi import FastAPI app = FastAPI( title="Nome do Sistema", version="1.0.0", description="Descrição do sistema", docs_url="/swagger", # UI Swagger em /swagger redoc_url="/redoc", # UI ReDoc em /redoc openapi_url="/openapi.json", ) ``` A documentação é gerada ao vivo — **nenhum arquivo precisa ser escrito à mão**. O `tech-writer` exporta o `/openapi.json` para o arquivo `docs/openapi.yaml` versionado no repositório: ```bash docker compose run --rm backend python -c \ "import json, yaml; from app.main import app; \ open('docs/openapi.yaml','w').write(yaml.dump(app.openapi()))" ``` Docstring por endpoint (usada automaticamente no Swagger): ```python @router.post("/processos", status_code=201, response_model=ProcessoResponse, summary="Criar processo", tags=["processos"]) async def criar_processo(body: CriarProcessoRequest, ...): """ Cria um novo processo administrativo. - **descricao**: texto descritivo do processo - **responsavel_id**: UUID do usuário responsável """ ``` ## Observabilidade: Prometheus + endpoints de saúde Use **`prometheus-fastapi-instrumentator`** — registra middleware global automático, sem precisar anotar cada endpoint: ```python # main.py from prometheus_fastapi_instrumentator import Instrumentator app = FastAPI(...) Instrumentator().instrument(app).expose(app, endpoint="/metrics") ``` **Endpoints de saúde** (obrigatórios): ```python @app.get("/healthz", tags=["health"], include_in_schema=False) async def healthz(): """Liveness probe — o processo está vivo.""" return {"status": "ok"} @app.get("/readyz", tags=["health"], include_in_schema=False) async def readyz(conn: AsyncConnection = Depends(get_conn)): """Readiness probe — dependências (banco, Redis) estão acessíveis.""" try: await conn.execute(text("SELECT 1")) return {"status": "ok"} except Exception: raise HTTPException(status_code=503, detail="banco indisponível") ``` ## Endpoint /version (obrigatório desde o primeiro commit) ```python # adapter/http/version_router.py import os from fastapi import APIRouter router = APIRouter() @router.get("/version", include_in_schema=False) async def version(): return { "app": os.getenv("APP_NAME", ""), "backend": os.getenv("APP_VERSION", ""), "frontend": os.getenv("FRONTEND_VERSION", ""), "buildDate": os.getenv("BUILD_DATE", ""), "commit": os.getenv("GIT_COMMIT", ""), } ``` ## Testes ```bash docker compose run --rm backend pytest --cov=app ``` Use `pytest-asyncio` para use cases assíncronos. Mock os ports via `unittest.mock.AsyncMock` ou implementação in-memory. ## APP_ENV e CORS ```python APP_ENV = os.getenv("APP_ENV", "production") @app.middleware("http") async def auth_middleware(request: Request, call_next): if APP_ENV == "development": return await call_next(request) # validar JWT... ``` ## Dois Dockerfiles - `Dockerfile` — produção: multi-stage, imagem final `python:3.12-slim`. - `Dockerfile.dev` — dev: instala `watchfiles` ou `uvicorn --reload`.