Aggregate 6 - Functional style

This example shows how to define and use your own immutable aggregate base class with a more “functional” style than example 5.

Base classes

The DomainEvent class is defined as a “frozen” Python dataclass.

@dataclass(frozen=True)
class DomainEvent:
    originator_id: UUID
    originator_version: int
    timestamp: datetime

The Aggregate base class in this example is also defined as a “frozen” Python dataclass.

@dataclass(frozen=True)
class Aggregate:
    id: UUID
    version: int
    created_on: datetime
    modified_on: datetime

The Snapshot class in this example is also defined as a “frozen” Python dataclass that extends DomainEvent.

@dataclass(frozen=True)
class Snapshot(DomainEvent):
    state: dict[str, Any]

    @classmethod
    def take(cls, aggregate: Aggregate) -> Snapshot:
        return Snapshot(
            originator_id=aggregate.id,
            originator_version=aggregate.version,
            timestamp=datetime_now_with_tzinfo(),
            state=aggregate.__dict__,
        )

A generic aggregate_projector function is also defined, which takes a mutator function and returns a function that can reconstruct an aggregate of a particular type from an iterable of domain events.

def aggregate_projector(
    mutator: MutatorFunction[TAggregate],
) -> Callable[[TAggregate | None, Iterable[DomainEvent]], TAggregate | None]:
    def project_aggregate(
        aggregate: TAggregate | None, events: Iterable[DomainEvent]
    ) -> TAggregate | None:
        for event in events:
            aggregate = mutator(event, aggregate)
        return aggregate

    return project_aggregate

Domain model

The Dog aggregate class is defined as immutable frozen data class that extends the aggregate base class.

@dataclass(frozen=True)
class Dog(Aggregate):
    name: str
    tricks: tuple[str, ...]

The aggregate event classes, DogRegistered and TrickAdded, are explicitly defined as separate module level classes.

@dataclass(frozen=True)
class DogRegistered(DomainEvent):
    name: str

@dataclass(frozen=True)
class TrickAdded(DomainEvent):
    trick: str

The aggregate commands, register_dog() and add_trick() are defined as module level functions.

def register_dog(name: str) -> DomainEvent:
    return DogRegistered(
        originator_id=uuid4(),
        originator_version=1,
        timestamp=datetime_now_with_tzinfo(),
        name=name,
    )

def add_trick(dog: Dog, trick: str) -> DomainEvent:
    return TrickAdded(
        originator_id=dog.id,
        originator_version=dog.version + 1,
        timestamp=datetime_now_with_tzinfo(),
        trick=trick,
    )

The mutator function, mutate_dog(), is defined as a module level function.

@singledispatch
def mutate_dog(_: DomainEvent, __: Dog | None) -> Dog | None:
    """Mutates aggregate with event."""


@mutate_dog.register
def _(event: DogRegistered, _: None) -> Dog:
    return Dog(
        id=event.originator_id,
        version=event.originator_version,
        created_on=event.timestamp,
        modified_on=event.timestamp,
        name=event.name,
        tricks=(),
    )


@mutate_dog.register
def _(event: TrickAdded, dog: Dog) -> Dog:
    return Dog(
        id=dog.id,
        version=event.originator_version,
        created_on=dog.created_on,
        modified_on=event.timestamp,
        name=dog.name,
        tricks=(*dog.tricks, event.trick),
    )


@mutate_dog.register
def _(event: Snapshot, _: None) -> Dog:
    return Dog(
        id=event.state["id"],
        version=event.state["version"],
        created_on=event.state["created_on"],
        modified_on=event.state["modified_on"],
        name=event.state["name"],
        tricks=tuple(event.state["tricks"]),  # comes back from JSON as a list
    )

The aggregate projector function, project_dog(), is defined as a module level function by calling aggregate_projector() with mutate_dog() as the argument.

project_dog = aggregate_projector(mutate_dog)

Application

The DogSchool application in this example uses the library’s Application class. It must receive the new events that are returned by the aggregate command methods, and pass them to its save() method. The aggregate projector function must also be supplied when reconstructing an aggregate from the repository, and when taking snapshots.

class DogSchool(Application[UUID]):
    is_snapshotting_enabled = True
    snapshot_class = Snapshot

    def register_dog(self, name: str) -> UUID:
        event = register_dog(name)
        self.save(event)
        return event.originator_id

    def add_trick(self, dog_id: UUID, trick: str) -> None:
        dog = self.repository.get(dog_id, projector_func=project_dog)
        self.save(add_trick(dog, trick))

    def get_dog(self, dog_id: UUID) -> dict[str, Any]:
        dog = self.repository.get(dog_id, projector_func=project_dog)
        return {"name": dog.name, "tricks": dog.tricks}

Test case

The TestDogSchool test case shows how the DogSchool application can be used.

class TestDogSchool(TestCase):
    def test_dog_school(self) -> None:
        # Construct application object.
        school = DogSchool()

        # Evolve application state.
        dog_id = school.register_dog("Fido")
        school.add_trick(dog_id, "roll over")
        school.add_trick(dog_id, "play dead")

        # Query application state.
        dog = school.get_dog(dog_id)
        assert dog["name"] == "Fido"
        assert dog["tricks"] == ("roll over", "play dead")

        # Select notifications.
        notifications = school.notification_log.select(start=1, limit=10)
        assert len(notifications) == 3

        # Take snapshot.
        school.take_snapshot(dog_id, version=3, projector_func=project_dog)
        dog = school.get_dog(dog_id)
        assert dog["name"] == "Fido"
        assert dog["tricks"] == ("roll over", "play dead")

        # Continue with snapshotted aggregate.
        school.add_trick(dog_id, "fetch ball")
        dog = school.get_dog(dog_id)
        assert dog["name"] == "Fido"
        assert dog["tricks"] == ("roll over", "play dead", "fetch ball")

Code reference

class examples.aggregate6.baseclasses.DomainEvent(originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime')

Bases: object

originator_id: UUID

originator_version: int

timestamp: datetime

class examples.aggregate6.baseclasses.Aggregate(id: 'UUID', version: 'int', created_on: 'datetime', modified_on: 'datetime')

Bases: object

id: UUID

version: int

created_on: datetime

modified_on: datetime

class examples.aggregate6.baseclasses.Snapshot(originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', state: 'dict[str, Any]')

Bases: DomainEvent

state: dict[str, Any]

classmethod take(aggregate: Aggregate) → Snapshot

examples.aggregate6.baseclasses.aggregate_projector(mutator: MutatorFunction[TAggregate]) → Callable[[TAggregate | None, Iterable[DomainEvent]], TAggregate | None]

class examples.aggregate6.domainmodel.Dog(id: 'UUID', version: 'int', created_on: 'datetime', modified_on: 'datetime', name: 'str', tricks: 'tuple[str, ...]')

Bases: Aggregate

name: str

tricks: tuple[str, ...]

class examples.aggregate6.domainmodel.DogRegistered(originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', name: 'str')

Bases: DomainEvent

name: str

class examples.aggregate6.domainmodel.TrickAdded(originator_id: 'UUID', originator_version: 'int', timestamp: 'datetime', trick: 'str')

Bases: DomainEvent

trick: str

examples.aggregate6.domainmodel.register_dog(name: str) → DomainEvent

examples.aggregate6.domainmodel.add_trick(dog: Dog, trick: str) → DomainEvent

examples.aggregate6.domainmodel.mutate_dog(_: DomainEvent, __: Dog | None) → Dog | None

examples.aggregate6.domainmodel.mutate_dog(event: DogRegistered, _: None) → Dog

examples.aggregate6.domainmodel.mutate_dog(event: TrickAdded, dog: Dog) → Dog

examples.aggregate6.domainmodel.mutate_dog(event: Snapshot, _: None) → Dog

Mutates aggregate with event.

class examples.aggregate6.application.DogSchool(env: Mapping[str, str] | None = None)

Bases: Application[UUID]

is_snapshotting_enabled: bool = True

snapshot_class

alias of Snapshot

register_dog(name: str) → UUID

add_trick(dog_id: UUID, trick: str) → None

get_dog(dog_id: UUID) → dict[str, Any]

name = 'DogSchool'

class examples.aggregate6.test_application.TestDogSchool(methodName='runTest')

Bases: TestCase

test_dog_school() → None