Unit of Work y Transacciones
El UoW encapsula el límite transaccional: un aggregate, una transacción, un puerto.
El patrón Unit of Work (UoW) define un límite transaccional: todas las operaciones que ocurren dentro de una unidad de trabajo se confirman juntas o se revierten juntas. En DDD, ese límite natural es el aggregate: una transacción modifica un único aggregate.
En la arquitectura hexagonal, el UoW es un puerto outbound (una abstracción ABC). La capa de aplicación lo usa como context manager: el handler abre la unidad de trabajo, realiza las operaciones de dominio, y al salir del bloque la implementación decide si hace commit o rollback. La aplicación no sabe si el backend es SQLAlchemy, MongoDB o una base de datos in-memory.
La tentación más común es manejar la transacción directamente en el repositorio o en el handler usando Session.commit(). Esto acopla la capa de aplicación a SQLAlchemy y hace imposible probar la transaccionalidad sin una base de datos real. El UoW como puerto resuelve ambos problemas.
Cuando múltiples repositorios deben participar en la misma transacción (por ejemplo, guardar el aggregate y emitir un outbox event en la misma operación atómica), el UoW puede coordinarlo internamente. Desde el handler, la vista sigue siendo simple: un bloque `with self._uow`.
El límite transaccional correcto en DDD es siempre un aggregate. Si un handler necesita modificar dos aggregates en la misma transacción, es una señal de que los límites del aggregate están mal definidos o de que se necesita consistencia eventual via domain events.
# shared_kernel/application/unit_of_work.py — Puerto (abstracción)
from abc import ABC, abstractmethod
from types import TracebackType
from typing import Type
class UnitOfWork(ABC):
"""
Puerto outbound: encapsula el límite transaccional.
La implementación concreta vive en infrastructure/.
"""
@abstractmethod
def __enter__(self) -> 'UnitOfWork': ...
@abstractmethod
def __exit__(
self,
exc_type: Type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> bool: ...
# __exit__ hace commit si exc_type is None, rollback si hay excepción
@abstractmethod
def commit(self) -> None: ...
@abstractmethod
def rollback(self) -> None: ...
# --- Uso en el Command Handler ---
class <DoSomething>Handler:
def __init__(self, repo: <Aggregate>Repository, uow: UnitOfWork, bus: EventBus) -> None:
self._repo = repo
self._uow = uow
self._bus = bus
def handle(self, command: <DoSomething>Command) -> None:
aggregate = self._repo.get_by_id(<Aggregate>Id(command.aggregate_id))
aggregate.<do_something>()
with self._uow: # abre transacción
self._repo.save(aggregate) # acumula cambios
# commit automático al salir del with (si no hubo excepción)
for event in aggregate.pull_events(): # fuera de la transacción
self._bus.publish(event)
# --- Implementación fake (para tests sin base de datos) ---
# tests/fakes/fake_unit_of_work.py
class FakeUnitOfWork(UnitOfWork):
def __init__(self) -> None:
self.committed = False
self.rolled_back = False
def __enter__(self) -> 'FakeUnitOfWork':
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
if exc_type is None:
self.commit()
else:
self.rollback()
return False # no suprimir excepciones
def commit(self) -> None: self.committed = True
def rollback(self) -> None: self.rolled_back = True
# --- Implementación concreta (SQLAlchemy) — vive en infrastructure/ ---
# infrastructure/persistence/sql_unit_of_work.py
from sqlalchemy.orm import Session
class SqlUnitOfWork(UnitOfWork):
def __init__(self, session_factory) -> None:
self._session_factory = session_factory
def __enter__(self) -> 'SqlUnitOfWork':
self._session: Session = self._session_factory()
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
if exc_type is None:
self.commit()
else:
self.rollback()
self._session.close()
return False
def commit(self) -> None: self._session.commit()
def rollback(self) -> None: self._session.rollback()Debugging lab
Detecta y corrige el error o la violación de diseño.
- 5.7.5.1
from sqlalchemy.orm import Session class PlaceOrderHandler: def __init__(self, session: Session): self._session = session def handle(self, cmd): order = Order.create(...) self._session.add(order) self._session.commit()
- 5.7.5.2
def handle(self, cmd): order = self._order_repo.get_by_id(OrderId(cmd.order_id)) customer = self._customer_repo.get_by_id(CustomerId(cmd.customer_id)) with self._uow: order.assign(CustomerId(cmd.customer_id)) customer.add_order_count() self._order_repo.save(order) self._customer_repo.save(customer)
- 5.7.5.3
class CancelOrderHandler: def handle(self, cmd): order = self._repo.get_by_id(OrderId(cmd.order_id)) with self._uow: order.cancel() self._repo.save(order) for e in order.pull_events(): self._bus.publish(e) # publicar dentro del with
- 5.7.5.4
class FakeUnitOfWork(UnitOfWork): def __enter__(self): return self def __exit__(self, *args): self.committed = True # siempre commit
- 5.7.5.5
class UpdateOrderHandler: def handle(self, cmd): order = self._repo.get_by_id(OrderId(cmd.order_id)) order.update_total(Money(cmd.new_total)) self._repo.save(order) # sin UoW — sin bloque with