Domain building blocks

Urich provides base types for the domain layer: Entity, ValueObject, AggregateRoot, DomainEvent, Repository, and EventBus. Import from urich.domain.

---

Entity

Identity-bearing object: equality and hash by id.

from urich.domain import Entity

class Order(Entity):
    def __init__(self, id: str, customer_id: str):
        super().__init__(id=id)
        self.customer_id = customer_id

---

ValueObject

Value without identity; equality by all fields. Uses a frozen dataclass.

from urich.domain import ValueObject
from dataclasses import dataclass

@dataclass(frozen=True)
class Money(ValueObject):
    amount_cents: int
    currency: str

---

AggregateRoot

Extends Entity. Holds a list of pending domain events that are raised during work and collected when the aggregate is saved.

from urich.domain import AggregateRoot, DomainEvent
from dataclasses import dataclass

@dataclass
class OrderCreated(DomainEvent):
    order_id: str
    customer_id: str
    total_cents: int

class Order(AggregateRoot):
    def __init__(self, id: str, customer_id: str, total_cents: int):
        super().__init__(id=id)
        self.customer_id = customer_id
        self.total_cents = total_cents
        self.raise_event(OrderCreated(order_id=id, customer_id=customer_id, total_cents=total_cents))

    # Later: order.collect_pending_events() returns the list and clears it

API:

• raise_event(event: DomainEvent) — Appends the event to the pending list.
• collect_pending_events() -> list[DomainEvent] — Returns the list and clears it. Typically called in the application layer after repo.add() or repo.save(), then each event is published to the EventBus.

---

DomainEvent

Base type for domain events. Subclass as dataclasses with fields.

from urich.domain import DomainEvent
from dataclasses import dataclass

@dataclass
class OrderCreated(DomainEvent):
    order_id: str
    customer_id: str
    total_cents: int

---

Repository

Abstract interface for aggregate persistence. Generic over the aggregate type.

from urich.domain import Repository
from typing import Optional

class IOrderRepository(Repository[Order]):
    pass

class OrderRepositoryImpl(IOrderRepository):
    async def get(self, id: str) -> Optional[Order]: ...
    async def add(self, aggregate: Order) -> None: ...
    async def save(self, aggregate: Order) -> None: ...

• get(id) — Load by id; return None if not found.
• add(aggregate) — Persist a new aggregate.
• save(aggregate) — Update an existing aggregate.

DomainModule registers the implementation in the container and resolves the interface to it so handlers get the repo by type.

---

EventBus

Protocol for publishing and subscribing to domain events. Provided by EventBusModule or by DomainModule (in-process) if none is registered.

from urich.domain.events import EventBus

# In a handler:
await self._event_bus.publish(OrderCreated(...))

Protocol:

• async def publish(self, event: DomainEvent) -> None
• def subscribe(self, event_type: type[DomainEvent], handler: Any) -> None

InProcessEventDispatcher is the default implementation: subscribe by event type, publish invokes all registered handlers (sync or async).

---

Imports summary

from urich.domain import (
    Entity,
    ValueObject,
    AggregateRoot,
    DomainEvent,
    Repository,
    EventBus,
    InProcessEventDispatcher,
)