Estructura Submodulos (por fases)¶
Usaremos Clean Architecture como base:
interface/web → application → domain ← infrastructure (las dependencias apuntan hacia el dominio).
Fase 1 — Esqueleto mínimo (carpetas base)¶
Crea esta estructura inicial (sin llenarla toda). La idea es empezar pequeño y crecer por fases:
submodule_name/
┣ domain/
┣ application/
┣ infrastructure/
┣ interface/
┃ ┗ web/
┣ docs/
┣ tests/
┣ __init__.py
┗ module.manifest.json
Qué va en cada una (resumen):
- domain/ → Reglas de negocio puras (entidades, invariantes, eventos).
- application/ → Casos de uso/servicios que orquestan el dominio.
- infrastructure/ → Implementaciones concretas (ORM, repos, APIs externas).
- interface/web/ → Blueprint Flask (rutas, vistas, guards/RBAC).
- docs/ → Documentación viva del submódulo.
- tests/ → Pruebas unitarias/integración.
- module.manifest.json → Metadatos del módulo (nombre, versión, eventos, deps).
Plantilla de module.manifest.json:
{
"name": "customer_management",
"version": "0.1.0",
"owners": ["Equipo CRM"],
"dependencies": ["core_auth"],
"events": {
"publishes": ["CustomerCreated", "CustomerUpdated"],
"subscribes": []
}
}
Fase 2 — Módulo mínimo funcional (MVP muy pequeño)¶
2.1 Dominio (la entidad mínima)¶
domain/customer_entity.py
from dataclasses import dataclass
@dataclass(frozen=True)
class Customer:
id: int | None
name: str
document: str
def validate(self):
if not self.name.strip():
raise ValueError("Customer.name vacío")
if not self.document.strip():
raise ValueError("Customer.document vacío")
2.2 Contrato de repositorio (puerto)¶
domain/contracts.py
from typing import Protocol, Iterable
from .customer_entity import Customer
class CustomerRepository(Protocol):
def create(self, c: Customer) -> Customer: ...
def get_by_id(self, id_: int) -> Customer | None: ...
def list(self) -> Iterable[Customer]: ...
2.3 Caso de uso (application)¶
application/customer_service.py
from ..domain.customer_entity import Customer
from ..domain.contracts import CustomerRepository
class CustomerService:
def __init__(self, repo: CustomerRepository):
self.repo = repo
def create_customer(self, name: str, document: str) -> Customer:
c = Customer(id=None, name=name, document=document)
c.validate()
return self.repo.create(c)
def list_customers(self):
return list(self.repo.list())
2.4 Web mínimo (Blueprint)¶
interface/web/web_bp.py
from flask import Blueprint, jsonify, request
from ...application.customer_service import CustomerService
# Inyecta un repo "falso" por ahora para el MVP
from ...infrastructure.repositories.memory_repo import InMemoryCustomerRepo
bp = Blueprint("customer_management", __name__, url_prefix="/customer_management")
_service = CustomerService(repo=InMemoryCustomerRepo())
@bp.get("/")
def health():
return jsonify({"ok": True, "module": "customer_management"})
@bp.post("/customers")
def create_customer():
data = request.get_json() or {}
c = _service.create_customer(data.get("name",""), data.get("document",""))
return jsonify({"id": c.id, "name": c.name, "document": c.document}), 201
@bp.get("/customers")
def list_customers():
customers = _service.list_customers()
return jsonify([{"id": c.id, "name": c.name, "document": c.document} for c in customers])
2.5 Repo en memoria (infra simple para empezar)¶
infrastructure/repositories/memory_repo.py
from ...domain.customer_entity import Customer
class InMemoryCustomerRepo:
def __init__(self):
self._data = []
self._seq = 1
def create(self, c: Customer) -> Customer:
new = Customer(id=self._seq, name=c.name, document=c.document)
self._seq += 1
self._data.append(new)
return new
def get_by_id(self, id_):
return next((x for x in self._data if x.id == id_), None)
def list(self):
return list(self._data)
✅ Con esto ya tienes rutas funcionales sin BD real. Ideal para integrar rápido al core y probar RBAC/guards.
Fase 3 — Documentación mínima viva¶
Crea estos archivos para que el módulo sea gobernable desde el día 1:
docs/
┣ index.md
┣ logic.md
┣ specifications.md
┗ external_interaction/
┣ integrations.md
┗ transactions/
┗ TX-01-create-customer.md
docs/index.md (plantilla breve)
# customer_management — Índice
- Visión: gestionar clientes (alta, edición, archivo).
- Interfaces: Blueprint `/customer_management`.
- Casos de uso: crear cliente, listar clientes.
- Estado: MVP sin persistencia real (repo en memoria).
docs/logic.md
# Lógica (resumen)
- Validaciones mínimas: `name`, `document` obligatorios.
- Invariantes en `domain/` (sin acceso a infraestructura).
- `application/` orquesta y lanza eventos (futuro).
docs/specifications.md
# Especificaciones
- Endpoint POST `/customers`: Crea cliente (201).
- Endpoint GET `/customers`: Lista clientes (200).
- RBAC: requiere `role:admin` (pendiente integrar guard).
docs/external_interaction/transactions/TX-01-create-customer.md
# TX-01: Create Customer
- Entrada: `name`, `document`.
- Reglas: no vacíos.
- Salida: `id`, `name`, `document`.
- Errores: 400 si validación falla.
Fase 4 — Persistencia real y límites externos¶
Cuando el MVP esté estable:
-
Modelo ORM
infrastructure/models/customer.pyfrom sqlalchemy.orm import Mapped, mapped_column, declarative_base Base = declarative_base() class CRM_CM_customer(Base): __tablename__ = "CRM_CM_customers" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) name: Mapped[str] document: Mapped[str] -
Repositorio SQLAlchemy
infrastructure/repositories/sqlalchemy_repo.pyfrom ...domain.customer_entity import Customer from ..models.customer import CRM_CM_customer from sqlalchemy.orm import Session class SqlAlchemyCustomerRepo: def __init__(self, session: Session): self.s = session def create(self, c: Customer) -> Customer: row = CRM_CM_customer(name=c.name, document=c.document) self.s.add(row); self.s.commit(); self.s.refresh(row) return Customer(id=row.id, name=row.name, document=row.document) def get_by_id(self, id_): row = self.s.get(CRM_CM_customer, id_) return None if not row else Customer(id=row.id, name=row.name, document=row.document) def list(self): return [Customer(id=r.id, name=r.name, document=r.document) for r in self.s.query(CRM_CM_customer).all()] -
Boundaries externos (si aplica)
infrastructure/boundaries/...para APIs externas (Facturama, Nubefact, etc.).
Mantén adaptadores delgados y mapea DTO ↔ dominio enapplication.
Fase 5 — Web/UI y guards¶
Estructura mínima para vistas/estáticos si usas HTML:
interface/web/
┣ blueprints/
┃ ┗ base/
┃ ┣ __init__.py
┃ ┗ base_bp.py
┣ templates/
┃ ┗ submodule_name/
┃ ┣ _layout.html
┃ ┗ index.html
┣ static/
┃ ┗ submodule_name/
┃ ┣ css/
┃ ┗ js/
┣ guards.py
┗ web_bp.py
guards.py: verifica RBAC/tenant antes de ejecutar rutas._layout.html: layout base del submódulo.index.html: landing simple del módulo (lista rápida, enlaces a TX).
Fase 6 — Pruebas¶
- Unitarias: dominio y application sin tocar BD.
- Integración: repos con DB (usa una SQLite temporal o transacciones rollback).
- Web: prueba de endpoints (Flask test client).
Estructura sugerida:
tests/
┣ test_domain_customer.py
┣ test_application_customer_service.py
┗ test_web_endpoints.py
Convenciones (rápidas)¶
- Prefijo de rutas:
/<submodule_name>(ej./customer_management). - Nombres:
customer_entity.py,customer_service.py,sqlalchemy_repo.py. - Dependencias hacia adentro (Clean):
interface/web→application→domain←infrastructure. - Documenta cada TX en
docs/external_interaction/transactions/.
Checklist por submódulo¶
- Fase 1: Esqueleto base creado.
- Fase 2: MVP (entidad + servicio + blueprint + repo en memoria).
- Fase 3: Docs mínimas (
index,logic,specifications,TX-01). - Fase 4: Persistencia real (ORM + repo SQLA).
- Fase 5: UI/guards listos (RBAC/tenant).
- Fase 6: Pruebas (dominio/app/web/infra).