Persistencia: ORM y Mappers
Los modelos ORM son una vista de la base de datos, no el dominio.
El modelo ORM y el modelo de dominio resuelven problemas distintos. El modelo ORM describe cómo los datos se almacenan en tablas relacionales: nombres de columnas, tipos SQL, relaciones de foreign key, índices. El modelo de dominio describe el comportamiento del negocio: invariantes, eventos, métodos con intención. Intentar que un solo objeto sirva a ambos propósitos produce una clase que no es buena en ninguno de los dos: termina siendo un Aggregate que no puede proteger invariantes porque SQLAlchemy necesita setters, o un ORM model lleno de lógica de negocio que hace los tests difíciles.
SQLAlchemy actúa como adapter de persistencia en la arquitectura hexagonal. El `Session` es el mecanismo concreto que gestiona la transacción y la identidad map; los modelos declarativos son el mapeo entre objetos Python y filas de tabla. Toda esta maquinaria vive en la capa de infraestructura. El dominio no la ve. Si mañana migras de SQLAlchemy a otro ORM o a un cliente directo de base de datos, el dominio no cambia ni una línea.
El mapper es la función —o par de funciones— que traduce entre el modelo de dominio y el modelo ORM. `to_domain(orm_row)` toma una instancia del ORM model y reconstruye el Aggregate con su estado completo, incluyendo objetos de valor embebidos y collections. `to_orm(aggregate)` hace el camino inverso: extrae el estado del Aggregate y lo vuelca en una instancia del ORM model lista para que SQLAlchemy la sincronice. El mapper es pura traducción: no aplica lógica de negocio, no ejecuta consultas, no abre sesiones.
Las invariantes del mapper son simples pero críticas: el dominio nunca importa el ORM; el ORM nunca llama métodos de dominio. El mapper puede importar ambos porque vive en infraestructura, pero su única responsabilidad es la traducción bidireccional. Si el mapper necesita consultar otro aggregate para reconstruir el dominio, algo está mal en el diseño del aggregate: probablemente está cargando demasiado en una sola reconstrucción.
La ventaja principal de esta separación es la estabilidad del dominio ante cambios de persistencia. Si el esquema de base de datos cambia —una columna se renombra, una tabla se normaliza— solo cambia el ORM model y el mapper. El Aggregate, los Value Objects y todas las reglas de negocio permanecen intactos. Del mismo modo, si la lógica de negocio evoluciona —nuevas invariantes, nuevos eventos— solo cambia el dominio; el mapper absorbe el delta de traducción sin contaminar el modelo relacional.
# infrastructure/persistence/models.py — ORM Models
# Only SQLAlchemy. No domain methods, no business rules.
from sqlalchemy import String, DateTime, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase): pass
class <AggregateORM>(Base):
__tablename__ = '<aggregates>'
id: Mapped[str] = mapped_column(String(36), primary_key=True)
<field_one>: Mapped[str] = mapped_column(String(200))
<field_two>: Mapped[str] = mapped_column(String(50))
# BAD: do NOT add domain methods here
# def <perform_action>(self): ... # <-- business logic in ORM model
# domain/model/<aggregate>.py — Domain Aggregate
# No SQLAlchemy. No persistence imports.
from dataclasses import dataclass, field
from uuid import UUID
@dataclass
class <Aggregate>:
id: UUID
_<field_one>: str
_<field_two>: <ValueObject>
_events: list = field(default_factory=list, repr=False)
def <perform_action>(self, <param>: <ValueObject>) -> None:
self._validate_<rule>(<param>)
self._<field_two> = <param>
self._events.append(<SomethingHappened>(aggregate_id=self.id))
# infrastructure/persistence/mappers.py — Translation only
# May import both domain and ORM. Must NOT run queries or apply business rules.
from ..models import <AggregateORM>
from ...domain.model.<aggregate> import <Aggregate>
from ...domain.model.<value_object> import <ValueObject>
def to_domain(row: <AggregateORM>) -> <Aggregate>:
return <Aggregate>(
id=row.id,
_<field_one>=row.<field_one>,
_<field_two>=<ValueObject>(row.<field_two>),
)
def to_orm(aggregate: <Aggregate>) -> <AggregateORM>:
return <AggregateORM>(
id=str(aggregate.id),
<field_one>=aggregate._<field_one>,
<field_two>=str(aggregate._<field_two>),
)Debugging lab
Detecta y corrige el error o la violación de diseño.
- 6.3.5.1
# domain/model/<aggregate>.py from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): pass class <Aggregate>(Base): __tablename__ = '<aggregates>' id: Mapped[str] = mapped_column(String(36), primary_key=True) <field>: Mapped[str] = mapped_column(String(200)) def <perform_action>(self, value: str) -> None: self.<field> = value
- 6.3.5.2
# infrastructure/persistence/mappers.py def to_domain(row: <AggregateORM>) -> <Aggregate>: aggregate = <Aggregate>(id=row.id, _<field>=row.<field>) if row.<status_field> == 'pending': aggregate.auto_approve() # business rule in mapper return aggregate
- 6.3.5.3
# infrastructure/persistence/models.py from ...domain.model.<aggregate> import <Aggregate> class <AggregateORM>(Base): __tablename__ = '<aggregates>' id: Mapped[str] = mapped_column(String(36), primary_key=True) def to_aggregate(self) -> <Aggregate>: # domain method in ORM model return <Aggregate>(id=UUID(self.id)) def apply_<rule>(self) -> None: # business logic in ORM model self.<field> = '<computed_value>'
- 6.3.5.4
# infrastructure/persistence/sql_<aggregate>_repository.py class Sql<Aggregate>Repository(<Aggregate>Repository): def get(self, aggregate_id: <AggregateId>) -> <AggregateORM>: # returns ORM return self._session.get(<AggregateORM>, str(aggregate_id)) def list_active(self) -> list[<AggregateORM>]: # returns ORM list return self._session.query(<AggregateORM>).filter_by(<status>='active').all()
- 6.3.5.5
# domain/model/<aggregate>.py from sqlalchemy import Column, String, event as sa_event from sqlalchemy.orm import relationship class <Aggregate>: _domain_events = relationship('<DomainEventORM>', backref='aggregate') def pull_events(self): return self._domain_events # returns ORM relationship proxy