Enum

Overview of Event Sourcing Most applications function by overwriting state. When a player picks up a sword in a game, a database record changes from two to three. This approach is efficient but destructive; it discards the history of how that state was reached. Event Sourcing flips this paradigm. Instead of storing the final balance or the current inventory count, you store a sequence of immutable events—facts that have happened in the past. By replaying these events from the beginning, you can reconstruct the state at any point in time. This provides an inherent audit log, simplifies debugging by allowing "time travel," and enables the creation of multiple "projections" or views of the same data without altering the source of truth. It is the same fundamental logic that powers Git and Blockchain. Prerequisites To follow this implementation, you should have a solid grasp of Python 3.10+ fundamentals, specifically **Object-Oriented Programming (OOP)**. Familiarity with Dataclasses and Enums is essential, as these provide the structure for immutable events. A basic understanding of **Dependency Injection** will also help when connecting the inventory logic to the underlying event store. Key Libraries & Tools - **enum**: Used to define distinct, readable event types like `ITEM_ADDED` and `ITEM_REMOVED`. - **dataclasses**: Provides a concise way to create event objects, specifically using `frozen=True` to ensure immutability. - **collections.Counter**: A specialized dictionary subclass for counting hashable objects, used here to aggregate inventory totals. - **functools.cache**: Implements a simple memoization strategy to avoid replaying the entire event history on every read request. - **Flox**: A tool for creating reproducible development environments, ensuring consistent package management across different machines. Code Walkthrough: Building the Core System Step 1: Defining Immutable Events We start by defining what an event looks like. It must contain the type of action and the data associated with it. ```python from dataclasses import dataclass, field from datetime import datetime from enum import Enum, auto class EventType(Enum): ITEM_ADDED = auto() ITEM_REMOVED = auto() @dataclass(frozen=True) class Event: type: EventType data: str timestamp: datetime = field(default_factory=datetime.now) ``` The `frozen=True` parameter is vital. Events represent the past; they cannot be changed once they occur. We use a `default_factory` for the timestamp to ensure each event is accurately placed in the timeline. Step 2: The Event Store and Caching The Event Store is a simple append-only list. To prevent performance degradation as the list grows, we apply a cache to the state reconstruction method. ```python from functools import cache from collections import Counter class Inventory: def __init__(self, store): self.store = store @cache def get_items(self): counts = Counter() for event in self.store.get_all_events(): if event.type == EventType.ITEM_ADDED: counts[event.data] += 1 elif event.type == EventType.ITEM_REMOVED: counts[event.data] -= 1 return {k: v for k, v in counts.items() if v > 0} def _invalidate_cache(self): self.get_items.cache_clear() ``` When we add an item, we append an event to the store and trigger `_invalidate_cache()`. The next time `get_items()` is called, it recalculates and recaches the state. Step 3: Advanced Projections Projections allow us to ask different questions of our data. For example, we can determine which items were collected most frequently, regardless of whether they are still in the inventory. ```python def get_most_collected(store): events = store.get_all_events() added_items = [e.data for e in events if e.type == EventType.ITEM_ADDED] return Counter(added_items).most_common(3) ``` Syntax Notes This implementation relies on **Generic Type Variables (`TypeVar`)** when evolving the system to handle complex objects rather than just strings. Using `typing.Generic[T]` allows the `Event` and `EventStore` classes to remain flexible, supporting any data structure while maintaining type safety. The use of the **decorator pattern** via `@cache` demonstrates a clean way to separate performance concerns from business logic. Practical Examples - **Financial Systems**: Storing every transaction (credit/debit) instead of just the balance to provide a perfect audit trail. - **E-commerce**: Tracking how long items sit in a cart before being removed to analyze user hesitation. - **Gaming**: Building a replay system by storing player inputs as events to recreate the match exactly. Tips & Gotchas - **Schema Evolution**: If you change the structure of your `Item` object later, your old events might break. You must plan for "upcasting" (transforming old events into the new format) or versioning your event schemas. - **Snapshotting**: For systems with millions of events, replaying from zero is too slow even with local caching. Periodically save a "snapshot" of the state so you only have to replay events from the last snapshot forward. - **Avoid for CRUD**: If your application only requires basic create, read, update, and delete operations without any need for history, event sourcing will introduce unnecessary complexity.