Saltar a contenido

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:

  1. Modelo ORM
    infrastructure/models/customer.py

    from 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]
    

  2. Repositorio SQLAlchemy
    infrastructure/repositories/sqlalchemy_repo.py

    from ...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()]
    

  3. Boundaries externos (si aplica)
    infrastructure/boundaries/... para APIs externas (Facturama, Nubefact, etc.).
    Mantén adaptadores delgados y mapea DTO ↔ dominio en application.


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/webapplicationdomaininfrastructure.
  • 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).