Skip to content

Domain (DDD building blocks)

pyfly.domain ships the foundational types every domain-driven design codebase ends up reinventing — Entity, ValueObject, AggregateRoot, DomainEvent, Specification, a DomainRepository protocol, and domain-flavoured exceptions. The module is pure standard-library Python with zero runtime dependencies, so you can import it from any layer of the application.

It mirrors org.fireflyframework.starter.domain (Java) and FireflyFramework.Starter.Domain (.NET) — but expressed natively in modern Python (Protocol, dataclass, StrEnum, Generic[T]).

Entity — identity-based equality

An entity is something whose identity is more important than its state. Two entities are equal iff they share the same id, regardless of any other field. Entities with id=None are transient (newly created, not yet persisted) and compare equal only by Python id().

from pyfly.domain import Entity

class Account(Entity[int]):
    def __init__(self, id: int | None = None, balance: int = 0) -> None:
        super().__init__(id)
        self.balance = balance

assert Account(id=1, balance=100) == Account(id=1, balance=999)  # same identity
assert Account(id=1) != Account(id=2)                            # different identity
assert Account() != Account()                                    # both transient

Hashing follows the same contract: persisted entities hash by (type, id), transient entities use object.__hash__.

ValueObject — structural equality, immutability

A value object is fully described by its attribute values. Subclass ValueObject and decorate with @dataclass(frozen=True) (or frozen=True, slots=True for performance). The base class is a marker for static analysis and adds a uniform replace(...) helper.

from dataclasses import dataclass
from pyfly.domain import ValueObject

@dataclass(frozen=True, slots=True)
class Money(ValueObject):
    amount: int
    currency: str

assert Money(100, "EUR") == Money(100, "EUR")
doubled = Money(100, "EUR").replace(amount=200)  # returns a new instance

Mutation raises dataclasses.FrozenInstanceError.

AggregateRoot — entity that owns a consistency boundary

An aggregate root extends Entity[TID] with a buffer of pending domain events. State changes happen through intent-revealing methods, which call self.raise_event(event) to queue an event. The events are published when the unit of work that persists the aggregate commits, never before: the EDA module's DomainEventPublisher collects the events an aggregate raises inside a unit of work (and those of an aggregate a unit saves) and publishes them as the unit commits, to the application's event listeners and, when configured, through the transactional outbox (see Domain events of aggregates). A unit that rolls back publishes nothing.

This is the non-event-sourced aggregate root. For the event-sourced variant (with apply/replay/when) see pyfly.eventsourcing.AggregateRoot.

from dataclasses import dataclass
from pyfly.container import service
from pyfly.data.transaction import transactional
from pyfly.domain import AggregateRoot, BusinessRuleViolation, DomainEvent

@dataclass(frozen=True)
class OrderShipped(DomainEvent):
    order_id: str = ""
    tracking_number: str = ""

class Order(AggregateRoot[str]):
    def __init__(self, id: str, status: str = "placed") -> None:
        super().__init__(id)
        self.status = status

    def ship(self, tracking_number: str) -> None:
        if self.status == "shipped":
            raise BusinessRuleViolation("order-already-shipped")
        self.status = "shipped"
        assert self.id is not None
        self.raise_event(OrderShipped(order_id=self.id, tracking_number=tracking_number))

@service
class Shipping:
    def __init__(self, orders: OrderRepository) -> None:
        self.orders = orders

    @transactional
    async def ship(self, order_id: str, tracking_number: str) -> None:
        order = await self.orders.find(order_id)
        order.ship(tracking_number)  # raises OrderShipped
        await self.orders.add(order)
        # OrderShipped is published as this unit commits: never if it rolls back

Do not publish the events yourself before the unit of work commits: a rollback would leave them published and the change undone. Code that publishes by hand drains them with clear_events() inside the unit and hands them to a publisher that joins it (DomainEventPublisher.publish(order), an outbox bus on the unit's datasource), or registers the publication with pyfly.data.transaction.after_commit.

An aggregate may also be an ORM-mapped entity (class Order(Base, AggregateRoot[int])): an instance the ORM loads without running AggregateRoot.__init__ still collects events.

Upgrading from 26.09.07: a behavior change. The EDA auto-configuration's DomainEventPublisher is on by default and drains the pending events as the unit commits. An application that collected pending_events() (or called clear_events()) after the unit and published them to the EDA bus now finds the buffer empty, and its integration events silently stop. Set pyfly.eda.domain-events.destination to have the publisher send them through the EDA bus (in the unit, with an outbox bus), or set pyfly.eda.domain-events.enabled: false to keep publishing by hand (see Domain events of aggregates).

DomainEvent — something that happened

DomainEvent is a frozen-dataclass base that auto-assigns a UUID event_id and a UTC occurred_at timestamp. The event_type property defaults to the subclass name.

from dataclasses import dataclass
from pyfly.domain import DomainEvent

@dataclass(frozen=True)
class CustomerRegistered(DomainEvent):
    customer_id: str = ""
    email: str = ""

evt = CustomerRegistered(customer_id="c-1", email="alice@example.com")
print(evt.event_id, evt.occurred_at, evt.event_type)

Every bus carries an event as JSON. to_payload() gives the event's fields as JSON values (an instant as ISO-8601 in UTC, a Decimal or UUID as a string, an Enum as its value, a nested dataclass as an object), and from_payload(payload) reads such a payload back into the typed event:

payload = evt.to_payload()
# {"event_id": "…", "occurred_at": "2026-09-27T10:30:15.250000+00:00",
#  "customer_id": "c-1", "email": "alice@example.com"}
assert CustomerRegistered.from_payload(payload) == evt

The CQRS command publisher and the domain-event publisher send this payload. Before 26.09.08 they sent dataclasses.asdict(event), whose datetime no bus could serialize, and the failure was only logged.

Specification — composable in-memory predicates

Specifications model business rules as objects: each one knows whether a given candidate satisfies it. Combine specs with & (and), | (or), and ~ (not).

from pyfly.domain import Specification

class IsAdult(Specification[Customer]):
    def is_satisfied_by(self, c: Customer) -> bool:
        return c.age >= 18

class IsPremium(Specification[Customer]):
    def is_satisfied_by(self, c: Customer) -> bool:
        return c.is_premium

# Compose:
eligible = IsAdult() & IsPremium()
fallback = ~IsAdult() | IsPremium()

# Use as a predicate (it's callable):
adults = list(filter(IsAdult(), customers))

# Or build one from a lambda:
spec = Specification.of(lambda c: c.balance > 1000, name="big-spender")

Note: pyfly.data.specification is a separate abstraction for backend-aware query predicates (it pushes the rule down into SQL or a MongoDB filter). The DDD specification is the in-memory predicate used inside aggregates and domain services. The two coexist.

DomainRepository — collection-like protocol

A DDD repository speaks in aggregates, not rows. The protocol is intentionally small — add, find, remove, next_id — because aggregate boundaries make complex queries unnecessary inside the domain. Read-side queries belong in projections and CQRS query handlers.

from pyfly.domain import AggregateRoot, DomainRepository

class Account(AggregateRoot[str]):
    ...

class AccountRepository(DomainRepository[Account, str]):
    ...

# A concrete implementation just needs the four methods. Because
# DomainRepository is a runtime-checkable Protocol, no inheritance is
# required.

Exceptions — DomainException / BusinessRuleViolation / AggregateNotFound

Domain exceptions extend pyfly.kernel.BusinessException so the existing RFC 7807 problem-details mapper, error filters, and @controller_advice handlers translate them automatically.

from pyfly.domain import BusinessRuleViolation, AggregateNotFound

raise BusinessRuleViolation(
    rule="loan-must-be-active",
    message="Cannot disburse a closed loan",
    context={"loan_id": loan_id},
)

raise AggregateNotFound("Loan", loan_id)

BusinessRuleViolation adds the rule name to context["rule"] and defaults the error code to DOMAIN_RULE_VIOLATION.

One-line bootstrap with the domain starter

For domain-tier microservices, decorate the application class with enable_domain_stack to activate CQRS, the transactional engine (saga/workflow/TCC), event sourcing, the rule engine, and the relational data layer (configure pyfly.data.relational.url: without it the start fails outside the dev profile, which falls back to ./app.db):

from pyfly.core import pyfly_application
from pyfly.starters.domain import enable_domain_stack

@enable_domain_stack
@pyfly_application(name="my-domain-service", scan_packages=["my_service"])
class Application:
    pass

The full DDD primitives are also re-exported from pyfly.starters.domain, so a single import line is enough:

from pyfly.starters.domain import (
    AggregateRoot, BusinessRuleViolation, DomainEvent,
    DomainRepository, Entity, Specification, ValueObject,
    enable_domain_stack,
)

See the samples/lumen/ directory for an end-to-end DDD service that uses every primitive on this page.