Factories
Creación compleja e invariante desde el inicio: construcción y reconstitución de aggregates.
Una Factory encapsula la lógica de creación de aggregates, entidades o Value Objects complejos cuando ese proceso no puede o no debe vivir en el constructor del propio objeto. Si construir un aggregate requiere múltiples pasos, validaciones cruzadas, cálculos o decisiones de negocio, es señal de que se necesita una Factory.
Las Factories tienen dos responsabilidades bien diferenciadas: creación (construir un aggregate nuevo desde datos de entrada, garantizando que nace con sus invariantes satisfechos) y reconstitución (reconstruir un aggregate desde datos persistidos, típicamente un modelo ORM o un evento, sin aplicar reglas de negocio duplicadas).
La separación entre creación y reconstitución es importante. Al crear, se validan las reglas de negocio y se genera una nueva identidad. Al reconstituir, el aggregate ya fue validado en su momento de creación; solo se rehidrata su estado desde el almacén de persistencia. Mezclar ambas operaciones en el mismo método produce código confuso y difícil de mantener.
Las Factories viven en la capa de dominio (`domain/factories/`). No deben llamar a repositorios, bases de datos ni servicios externos. Si necesitan información para construir el aggregate, la reciben como parámetros. La decisión de cuándo llamar a la Factory pertenece a la capa de aplicación.
En Python idiomático, las Factories se implementan como clases con métodos estáticos o de clase (`@staticmethod`, `@classmethod`), o como funciones libres. También es válido usar métodos de clase en el propio aggregate (`<Aggregate>.create(...)`) cuando la complejidad es moderada y no hay dependencias externas.
El test de una Factory es sencillo: construir un aggregate con datos válidos e inválidos, verificar invariantes, verificar identidad y verificar los Domain Events emitidos en la creación. Sin mocks, sin base de datos.
# domain/factories/<aggregate>_factory.py
# Factory: creación compleja e invariante. SIN acceso a BD ni repositorios.
# DOS responsabilidades: create() (nuevo) y reconstitute() (desde persistencia).
from dataclasses import dataclass, field
from uuid import UUID, uuid4
from typing import Any
from ..<model>.<aggregate> import <Aggregate>
from ..<model>.<value_object> import <ValueObject>
from ..events.<something_happened> import <SomethingHappened>
from ..errors import <DomainError>
class <Aggregate>Factory:
@staticmethod
def create(
<input_data>: <InputType>,
<another_param>: <Type>,
) -> <Aggregate>:
"""
Construye un nuevo <Aggregate> con identidad generada.
Valida invariantes de creación y emite el event de creación.
NO llama a repositorios ni a la base de datos.
"""
# 1. Validar reglas de negocio de creación
if not <some_business_rule>(<input_data>):
raise <DomainError>('<invariant_violated>')
# 2. Construir Value Objects desde datos crudos
<vo> = <ValueObject>.of(<input_data>.<field>)
# 3. Crear el aggregate con identidad nueva
aggregate = <Aggregate>(
id=uuid4(),
<vo>=<vo>,
<status>=<Status>.INITIAL,
)
# 4. Registrar el event de creación
aggregate._events.append(
<SomethingHappened>(aggregate_id=aggregate.id, ...)
)
return aggregate
@staticmethod
def reconstitute(
id: UUID,
<persisted_field>: <Type>,
<another_field>: <Type>,
) -> <Aggregate>:
"""
Rehidrata un <Aggregate> desde datos persistidos.
Sin validaciones de negocio (ya se validaron al crear).
Sin nuevos Domain Events.
"""
return <Aggregate>(
id=id,
<field>=<persisted_field>,
# Reconstruir VOs desde los valores almacenados
<vo>=<ValueObject>.of(<persisted_field>),
)
# ── Uso en Application Service ───────────────────────────────────────────────
# application/commands/<create_aggregate>/handler.py (fragmento)
class <CreateAggregateHandler>:
def handle(self, cmd: <CreateAggregateCommand>) -> UUID:
aggregate = <Aggregate>Factory.create(
<input_data>=cmd.<data>,
) # Factory crea; handler no sabe cómo construir
self._repo.save(aggregate)
events = aggregate.pull_events()
for event in events:
self._event_bus.publish(event)
return aggregate.idDebugging lab
Detecta y corrige el error o la violación de diseño.
- 4.4.5.1
# domain/factories/order_factory.py class OrderFactory: def __init__(self, repo: OrderRepository) -> None: self._repo = repo # ← error: repositorio en la Factory def create(self, customer_id: UUID, items: list[dict]) -> Order: existing = self._repo.find_by_customer(customer_id) # ← error: I/O en Factory ...
- 4.4.5.2
# infrastructure/persistence/mappers.py def to_domain(orm_model: OrderOrmModel) -> Order: return Order( id=orm_model.id, customer_id=orm_model.customer_id, items=[...], status=OrderStatus(orm_model.status), ) # ← usa constructor directo; no usa Factory.reconstitute()
- 4.4.5.3
# domain/factories/invoice_factory.py class InvoiceFactory: @staticmethod def create(order_id: UUID, amount: Decimal) -> Invoice: invoice = Invoice(id=uuid4(), order_id=order_id, amount=amount) return invoice # ← no emite Domain Event de creación @staticmethod def reconstitute(id: UUID, order_id: UUID, amount: Decimal) -> Invoice: invoice = Invoice(id=id, order_id=order_id, amount=amount) invoice._events.append(InvoiceCreated(invoice_id=id)) # ← error: event en reconstitute
- 4.4.5.4
# domain/factories/shipment_factory.py class ShipmentFactory: @staticmethod def create(order_id: UUID, address: str) -> Shipment: # ← address como str primitivo return Shipment(id=uuid4(), order_id=order_id, address=address)
- 4.4.5.5
# application/commands/place_order/handler.py class PlaceOrderHandler: def handle(self, cmd: PlaceOrderCommand) -> None: # ← construyendo el aggregate directamente en el handler order = Order( id=uuid4(), customer_id=cmd.customer_id, items=[OrderItem(product_id=i.product_id, qty=i.qty) for i in cmd.items], status=OrderStatus.PENDING, discount=Decimal('0'), ) self._repo.save(order)