pydomain — DDD, CQRS & Event Sourcing Building Blocks for Python

pydomain is a Python 3.12+ library that provides the tactical and architectural building blocks for building systems using Domain-Driven Design (DDD), Command-Query Responsibility Segregation (CQRS), and Event Sourcing (ES) — individually or composed together.

Design Philosophy

Opinionated about patterns, unopinionated about infrastructure.

pydomain gives you the abstractions, base classes, and wiring for DDD/CQRS/ES patterns. You provide the domain model and the storage adapters. The library never locks you into a specific database, message broker, or web framework.

This means:

• Domain layer has zero infrastructure imports. Your entities, value objects, and aggregates depend only on pydantic, uuid-utils, and the standard library. This holds for what the layer uses; until 0.4.0, importing pydomain.ddd still loads the other layers too (DDD-17, W2-11).
• Infrastructure is pluggable. The Repository and EventStore contracts are defined as typing.Protocol — any class with the right methods conforms. No inheritance required.
• You adopt incrementally. Use Level 1 (DDD only) for a rich domain model on top of a traditional ORM. Add CQRS, the message bus (with sagas and the outbox), and Event Sourcing as your system grows.

The Five Adoption Levels

You don't need all five. Start with what you need and add levels as your system grows.

Moving up a level adds capabilities without rewriting what already works. A ValueObject at Level 1 is the same class at Level 5. An AggregateRoot becomes an EventSourcedAggregateRoot by changing the base class — the command methods and invariants stay the same.

Built on Pydantic v2

Every domain concept (entities, value objects, events, commands, queries) is a Pydantic BaseModel. This gives you:

• Built-in validation — field constraints and custom validators express domain invariants directly in the type system.
• Serialization for free — model_dump() and model_validate() give round-trip serialization for event stores, message brokers, and API boundaries.
• Type safety — generic base classes like Entity[TId], Command[TResult], and Repository[T, TId] give you typed results with no casting.
• JSON Schema generation — self-documenting event and command catalogs for cross-team integration contracts.

Minimal Dependencies

pydomain has only two runtime dependencies:

Everything else — your database driver, web framework, message broker — is your choice.

Installation

pip install pydomain-lib

Requires Python 3.12+. Pulls in pydantic>=2.12.5 and uuid-utils>=0.10 automatically.

Upgrading? Read How to Upgrade to 0.3.1 first: it lists every behaviour change since 0.3.0, starting with the Pydantic floor.

For development tooling (pytest, ruff, mypy, pre-commit):

pip install "pydomain-lib[dev]"

To run the Protocol contract suites (pydomain.testing.contracts) against your own adapters, install the testing extra. It adds pytest>=8.0 and anyio>=4.0, which only the suites import:

pip install "pydomain-lib[testing]"

Note: The package is installed as pydomain-lib but imported as pydomain. All code examples below use import pydomain.

Quick Start — Your First Aggregate in 5 Minutes

from uuid import UUID
from pydomain.ddd.value_object import ValueObject
from pydomain.ddd.aggregate_root import AggregateRoot
from pydomain.ddd.domain_event import DomainEvent
from pydomain.testing.fake_repository import FakeRepository


# 1. Define a Value Object (immutable, defined by its attributes)
class Money(ValueObject):
    amount: int
    currency: str

    def add(self, other: "Money") -> "Money":
        if self.currency != other.currency:
            raise ValueError("Cannot add different currencies")
        return type(self)(amount=self.amount + other.amount, currency=self.currency)


# 2. Define a Domain Event (past-tense fact)
class OrderPlaced(DomainEvent):
    order_id: UUID
    total_amount: int
    currency: str


# 3. Define an Aggregate Root (consistency boundary, owns events)
class Order(AggregateRoot[UUID]):
    customer_id: UUID
    total: Money
    status: str = "pending"

    def place(self) -> None:
        if self.status != "pending":
            raise ValueError("Order is not pending")
        self.status = "placed"
        self._add_event(OrderPlaced(
            order_id=self.id,
            total_amount=self.total.amount,
            currency=self.total.currency,
        ))


# 4. Create and use it — id is auto-generated (UUIDv7)
order = Order(
    customer_id=UUID("550e8400-e29b-41d4-a716-446655440000"),
    total=Money(amount=1000, currency="EUR"),
)

order.place()                                 # records OrderPlaced event
events = order.pull_events()                  # collect recorded events
print(f"Status: {order.status}")              # "placed"


# 5. Test with fakes — no database needed
async def main():
    repo: FakeRepository[Order, UUID] = FakeRepository()
    await repo.save(order)
    found = await repo.get_by_id(order.id)
    assert found is not None and found.status == "placed"

Documentation

Development

git clone https://github.com/mgourlis/pydomain.git
cd pydomain

# Using pip (standard)
pip install -e ".[dev]"

# Or using uv (faster alternative)
uv sync --extra dev

The make targets call python -m …, so run them inside the activated environment or as uv run make test.

References

• Domain-Driven Design — Evans (2003)
• Patterns of Enterprise Application Architecture — Fowler (2002)
• Event Sourcing — Fowler
• CQRS — Fowler
• Diátaxis Documentation Framework

License

MIT License — see LICENSE file for details.