Aggregate 8 - Pydantic mutable

This example shows how to use Pydantic with the library’s declarative syntax.

Similar to example 1, aggregates are expressed using the library’s declarative syntax. This is the most concise way of defining an event-sourced aggregate.

Similar to example 7, domain event and custom value objects are defined using Pydantic. The main advantage of using Pydantic here is that any custom value objects used in the domain model will be automatically serialised and deserialised, without needing also to define custom transcoding classes.

Pydantic mutable model

The library’s eventsourcing.pydantic.mutablemodel defines base classes for aggregates that can use the library’s declarative syntax for mutable aggregates.

class Aggregate(BaseAggregate):
    @staticmethod
    def create_id(*_: Any, **__: Any) -> UUID:
        """Returns a new aggregate ID."""
        return uuid4()

    class Event(DomainEvent, CanMutateAggregate):
        pass

    class Created(Event, CanInitAggregate):
        originator_topic: str

class AggregateSnapshot(DomainEvent, CanSnapshotAggregate):
    topic: str
    state: Any

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        type_of_snapshot_state = typing.get_type_hints(cls)["state"]
        try:
            assert issubclass(
                type_of_snapshot_state, SnapshotState
            ), type_of_snapshot_state
        except (TypeError, AssertionError) as e:
            msg = (
                f"Subclass of {SnapshotState}"
                f" is required as the annotated type of 'state' on "
                f"{cls}, got: {type_of_snapshot_state}"
            )
            raise TypeError(msg) from e

class SnapshotState(Immutable):
    model_config = ConfigDict(extra="allow")

    def __init__(self, **kwargs: Any) -> None:
        for key in ["_created_on", "_modified_on"]:
            kwargs[key] = datetime_adapter.validate_python(kwargs[key])
        super().__init__(**kwargs)

Domain model

The code below shows how to define a mutable aggregate with the library’s declarative syntax, using the Pydantic mutable model.

from __future__ import annotations

from eventsourcing.domain import event
from eventsourcing.pydantic.immutablemodel import Immutable
from eventsourcing.pydantic.mutablemodel import (
    Aggregate,
    AggregateSnapshot,
    SnapshotState,
)


class Trick(Immutable):
    name: str


class DogSnapshotState(SnapshotState):
    name: str
    tricks: list[Trick]


class Dog(Aggregate):
    class Snapshot(AggregateSnapshot):
        state: DogSnapshotState

    @event("Registered")
    def __init__(self, name: str) -> None:
        self.name = name
        self.tricks: list[Trick] = []

    @event("TrickAdded")
    def add_trick(self, trick: Trick) -> None:
        self.tricks.append(trick)

Application

The DogSchool application in this example uses the library’s PydanticApplication class.

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from eventsourcing.pydantic.application import PydanticApplication
from examples.aggregate8.domainmodel import Dog, Trick

if TYPE_CHECKING:
    from uuid import UUID


class DogSchool(PydanticApplication):
    is_snapshotting_enabled = True

    def register_dog(self, name: str) -> UUID:
        dog = Dog(name)
        self.save(dog)
        return dog.id

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

    def get_dog(self, dog_id: UUID) -> dict[str, Any]:
        dog: Dog = self.repository.get(dog_id)
        return {
            "name": dog.name,
            "tricks": tuple([t.name for t in dog.tricks]),
            "created_on": dog.created_on,
            "modified_on": dog.modified_on,
        }

Test case

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

from __future__ import annotations

from datetime import datetime
from unittest import TestCase

from examples.aggregate8.application import DogSchool


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)
        self.assertEqual(dog["name"], "Fido")
        self.assertEqual(dog["tricks"], ("roll over", "play dead"))
        self.assertIsInstance(dog["created_on"], datetime)
        self.assertIsInstance(dog["modified_on"], datetime)

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

        # Take snapshot.
        school.take_snapshot(dog_id, version=3)
        dog = school.get_dog(dog_id)
        self.assertEqual(dog["name"], "Fido")
        self.assertEqual(dog["tricks"], ("roll over", "play dead"))
        self.assertIsInstance(dog["created_on"], datetime)
        self.assertIsInstance(dog["modified_on"], datetime)

        # Continue with snapshotted aggregate.
        school.add_trick(dog_id, "fetch ball")
        dog = school.get_dog(dog_id)
        self.assertEqual(dog["name"], "Fido")
        self.assertEqual(dog["tricks"], ("roll over", "play dead", "fetch ball"))
        self.assertIsInstance(dog["created_on"], datetime)
        self.assertIsInstance(dog["modified_on"], datetime)

Code reference

class eventsourcing.pydantic.mutablemodel.SnapshotState(**kwargs: Any)

Bases: Immutable

model_config = {'extra': 'allow', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class eventsourcing.pydantic.mutablemodel.AggregateSnapshot(*, originator_id: TAggregateID, originator_version: int, topic: str, state: Any, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>)

Bases: DomainEvent, CanSnapshotAggregate

topic: str

state: Any

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

originator_id_type

alias of UUID

class eventsourcing.pydantic.mutablemodel.Aggregate(*args: Any, **kwargs: Any)

Bases: BaseAggregate

static create_id(*_: Any, **__: Any) → UUID

Returns a new aggregate ID.

class Event(*, originator_id: TAggregateID, originator_version: int, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>)

Bases: DomainEvent, CanMutateAggregate

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

originator_id_type

alias of UUID

class Created(*, originator_id: TAggregateID, originator_version: int, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>, originator_topic: str)

Bases: Event, CanInitAggregate

originator_topic: str

String describing the path to an aggregate class.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

originator_id_type

alias of UUID

class examples.aggregate8.domainmodel.Trick(*, name: str)

Bases: Immutable

name: str

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class examples.aggregate8.domainmodel.DogSnapshotState(*, name: str, tricks: list[Trick], **kwargs: Any)

Bases: SnapshotState

name: str

tricks: list[Trick]

model_config = {'extra': 'allow', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class examples.aggregate8.domainmodel.Dog(*args: Any, **kwargs: Any)

Bases: Aggregate

class Snapshot(*, originator_id: TAggregateID, originator_version: int, topic: str, state: DogSnapshotState, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>)

Bases: AggregateSnapshot

state: DogSnapshotState

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

originator_id_type

alias of UUID

tricks: list[Trick]

add_trick(trick: Trick) → None

class Created(*, originator_id: TAggregateID, originator_version: int, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>, originator_topic: str)

Bases: Event, Created

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

originator_id_type

alias of UUID

class Event(*, originator_id: TAggregateID, originator_version: int, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>)

Bases: Event

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

originator_id_type

alias of UUID

class Registered(*, originator_id: TAggregateID, originator_version: int, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>, originator_topic: str, name: str)

Bases: Created, Event

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

originator_id_type

alias of UUID

name: str

class TrickAdded(*, originator_id: TAggregateID, originator_version: int, timestamp: datetime = <factory>, metadata: dict[str, str]=<factory>, event_id: UUID = <factory>, trick: Trick)

Bases: DecoratedFuncCaller, Event

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

trick: Trick

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

Bases: PydanticApplication

is_snapshotting_enabled: bool = True

register_dog(name: str) → UUID

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

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

aggregate_id_type

alias of UUID

name = 'DogSchool'

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

Bases: TestCase

test_dog_school() → None