eventsourcing

pyeventsourcing provides a comprehensive suite of tools for building robust event-sourced systems:

• Flexible Event Store: Extensible persistence using mappers and recorders. Supports custom transcoders and various database backends via environment variables.
• Security & Efficiency: Built-in application-level encryption (for GDPR/PII compliance) and compression (typically reducing event size by 25-50%).
• Performance: Snapshotting to reduce access time for aggregates with large event streams.
• Evolution: Versioning for both domain events and aggregate classes, allowing for upcasting of old data to new models.
• Reliability: Optimistic concurrency control for distributed environments and hash-chaining for data integrity verification.
• Observability: Correlation and causation IDs for tracing stories through multiple applications.
• CQRS Support: Notifications and projections for building materialized views and decoupled event-driven systems.

In pyeventsourcing, aggregate state is reconstructed by applying a sequence of events to an aggregate object. This is achieved through two primary methods on event classes:

1. mutate(self, aggregate: Any) -> Any: This method is responsible for the orchestration of state evolution. For 'subsequent' events, it validates the event against the current aggregate (checking originator_id and originator_version), calls apply(), increments the aggregate's version, and updates modified_on.
2. apply(self, aggregate: Any) -> None: This is the method you should override in your event subclasses to define the actual logic of how an event modifies the aggregate's attributes.

By iterating through a sequence of events and calling mutate() on each, you can implement an 'aggregate projector' to reconstruct the current state from history.

from typing import Any

class MyEvent(AggregateEvent):
    full_name: str

    def apply(self, aggregate: Any) -> None:
        aggregate.full_name = self.full_name

# Usage:
# a = my_event.mutate(a)
# assert a.full_name == "Eric Idle"

Since aggregate IDs cannot be changed once created, you can use Version-5 UUIDs to create 'Index Aggregates'. This allows you to identify an aggregate by a mutable attribute (like a name) even if the primary ID is a random UUID.

Pattern:

1. Generate a deterministic UUIDv5 based on a namespace and a name (e.g., /pages/{name}).
2. Create an Index aggregate using this deterministic ID.
3. Store the actual Page aggregate's UUID inside the Index aggregate.
4. To find a page by name: Recreate the UUIDv5 $\rightarrow$ Retrieve the Index aggregate $\rightarrow$ Get the Page ID $\rightarrow$ Retrieve the Page aggregate.

from uuid import NAMESPACE_URL, uuid5
from eventsourcing.domain import Aggregate

class Index(Aggregate):
    def __init__(self, name: str, ref: UUID):
        self.name = name
        self.ref = ref

    @classmethod
    def create(cls, name: str, ref: UUID) -> Self:
        return cls._create(
            event_class=cls.Created,
            id=cls.create_id(name),  # Deterministic ID
            name=name,
            ref=ref
        )

    @staticmethod
    def create_id(name: str) -> UUID:
        return uuid5(NAMESPACE_URL, f"/pages/{name}")

A Group extends Perspective and allows for decision-making that affects multiple EnduringObjects. The group's consistency boundary is the union of the boundaries of its members.

When a Group calls trigger_event(), the resulting decision is tagged with the continuity IDs of all members in the group. This allows a single event to evolve the state of multiple objects simultaneously (the 'one fact magic').

from eventsourcing.dcb.domain import Group

class StudentAndCourse(Group[Decision]):
    def __init__(self, student: Student | None, course: Course | None) -> None:
        self.student = student
        self.course = course

    def student_joins_course(self) -> None:
        # Business rules
        assert len(self.student.course_ids) < self.student.max_courses
        assert len(self.course.student_ids) < self.course.max_students
        
        # One event affects both members
        self.trigger_event(
            StudentJoinedCourse,
            student_id=self.student.id,
            course_id=self.course.id,
        )

# Usage
group = StudentAndCourse(student, course)
group.student_joins_course()