.ai-style-rules.md — DataFlow Analytics API

Generado por: /inherit-legacy-style · Modo: First-time Full-Scan · Tier: Small (38 source files) Commit fingerprint: a4f2c1d · Fecha: 2026-06-18 · Resuelto por: Álvaro (lead dev)

---

[Golden Files]

Los siguientes archivos son los ejemplares canónicos que el agente debe imitar. Cuando generes código nuevo, abre uno de estos primero y úsalo como plantilla estructural.

Archivo	Qué demuestra
api/routes/orders.py	Anatomía canónica de un router FastAPI: imports → schemas Pydantic → instancia router → endpoints ordenados GET/POST/PUT/DELETE → helpers privados al final
services/order_service.py	Patrón de servicio: clase con __init__(self, db: Session), métodos síncronos, DataFlowError para excepciones de negocio
api/middleware/logging.py	Única fuente de verdad del logger: get_logger(__name__) — nunca logging.getLogger() directamente
api/utils/formatters.py	Utilitarios: funciones puras sin estado, sin imports de modelos, sin side effects
models/order.py	Modelo SQLAlchemy: __tablename__, columnas con comentarios inline, relaciones al final del cuerpo

---

[Naming & State-Control Rules]

Logger

# CORRECTO — siempre
from api.middleware.logging import get_logger
logger = get_logger(__name__)

# PROHIBIDO
import logging
logger = logging.getLogger(__name__)  # DON'T

Type annotations

El proyecto usa PEP 604 (X | None) en módulos nuevos (Python 3.10+), pero los módulos legacy usan Optional[X]. Regla de convivencia acordada:

• En archivos models/ y services/: mantener Optional[X] hasta refactor explícito
• En archivos api/routes/ nuevos: usar X | None
• Nunca mezclar ambos en el mismo archivo

# api/routes/analytics.py (nuevo) — OK
def get_report(period: str | None = None) -> ReportSchema | None:

# models/order.py (legacy) — OK
from typing import Optional
def get_order(order_id: Optional[int] = None) -> Optional[Order]:

Async / Sync

Los servicios son completamente síncronos. Los endpoints FastAPI usan async def solo para I/O directo (Redis, llamadas HTTP externas). SQLAlchemy se llama de forma síncrona.

# CORRECTO
@router.get("/orders/{order_id}")
async def get_order(order_id: int, db: Session = Depends(get_db)):
    return order_service.get_by_id(db, order_id)  # servicio síncrono

# PROHIBIDO — no envuelvas servicios en await
result = await order_service.get_by_id(db, order_id)  # DON'T

Nomenclatura de variables de estado async (Celery tasks)

task_id       # UUID de la tarea Celery (str, nunca UUID object)
task_status   # Literal["pending", "running", "done", "failed"]
task_result   # dict | None, nunca un objeto Pydantic directamente

Errores de negocio

# Lanzar siempre DataFlowError (definido en api/utils/exceptions.py)
from api.utils.exceptions import DataFlowError

raise DataFlowError(code="ORDER_NOT_FOUND", detail=f"Order {order_id} no existe")

# PROHIBIDO — no lanzar HTTPException desde servicios
raise HTTPException(status_code=404, ...)  # DON'T — solo en routers, no en servicios

Paginación

# Parámetro estándar en todos los listados
def list_orders(db: Session, *, page: int = 1, page_size: int = 50) -> PaginatedResult:
    offset = (page - 1) * page_size
    ...

# PROHIBIDO — no usar skip/limit como nombres
def list_orders(db, skip=0, limit=50):  # DON'T

Imports — orden interno

1. stdlib
2. fastapi / pydantic / sqlalchemy
3. proyecto (api.*, models.*, services.*, workers.*)
4. — línea en blanco antes de los imports locales del mismo módulo —

---

[DONTs] — Anti-patrones que NO deben propagarse

#	Anti-patrón	Encontrado en	Motivo
1	logging.exception() o logging.getLogger()	analytics.py (legacy)	El logger custom añade trace_id al contexto; el estándar no
2	Optional[X] en archivos api/routes/ nuevos	products.py (conflicto junior)	Se resolvió: rutas nuevas → PEP 604
3	raise HTTPException desde capa de servicios	analytics_service.py línea 87	Rompe la separación de capas; el router decide el código HTTP
4	Crear módulos directamente en raíz del proyecto	Intento IA (sesión anterior)	Estructura fija: api/, services/, models/, workers/
5	async def en métodos de services/	Intento IA (sesión anterior)	SQLAlchemy 1.4 síncrono; la sesión no es thread-safe con await
6	skip / limit como parámetros de paginación	Ninguno aún	Prevención: el equipo acordó page + page_size
7	print() para debug	export_worker.py línea 34	Residuo legacy; no añadir más; usar logger.debug()

---

[Hooks instalados]

Soft hook (opción 1) — referencia en CLAUDE.md:

## Style Rules
@.ai-style-rules.md

El agente cargará estas reglas automáticamente al inicio de cada sesión. Antes de escribir cualquier código, abrirá el Golden File correspondiente y declarará qué ejemplar sigue y qué DONTs evita.

---

[Resumen de conflictos resueltos]

Conflicto	División detectada	Resolución del usuario
Type hints: Optional vs X | None	22 archivos Optional vs 8 archivos PEP 604	Convivencia por capa (ver sección Naming)
Error handling: try/except inline vs DataFlowError global	orders.py vs products.py	DataFlowError es el estándar; try/except solo en workers
Logger: get_logger vs logging.getLogger	35 vs 3 archivos	Suprimido automáticamente (señal débil 8%) — get_logger gana
Async en servicios: presente vs ausente	1 vs 37 archivos	Suprimido automáticamente — servicios síncronos

---

Próxima revisión sugerida: después de añadir el módulo billing/ (Sprint 14). Ejecutar /inherit-legacy-style en modo Branch B.