Python Event Upcasting for State Reconstruction

Hello! Welcome back to our module on State Management with Event Sourcing.

In our previous lesson, we explored the theoretical landscape of event schema evolution, discussing strategies ranging from run-time transformations like upcasting to full-scale data migrations. We established that upcasting is a common and balanced approach for handling non-breaking changes by transforming events on the fly.

Today, we move from theory to practice. This lesson is designed to meet the learning outcome: Implement an upcaster function in Python to transform an older event version into the current version during state reconstruction. We will take the concepts from our last discussion and turn them into working code, ensuring the domain model remains clean and only ever interacts with the latest version of any event.

1. The Mechanics of Upcasting

Let's quickly recap the core idea. Upcasting is a process that intercepts event data as it's read from the event store and transforms it from an older version to the current version before it is used to instantiate an event object and applied to an aggregate.

The key components are:

1. Event Versioning: Each event schema is assigned a version number. This is stored along with the event data.
2. Transformation Logic: A function or method exists for each required transformation (e.g., from v1 to v2, v2 to v3).
3. Execution on Read: The system (either the repository or a data mapper) detects a version mismatch between the stored event data and the current event class definition. It then invokes the appropriate transformation logic, potentially chaining multiple upcasters if needed (e.g., v1 -> v2 -> v3).

The aggregate's business logic is completely shielded from this process. It receives a fully-formed, current-version event object, as if all events in the store were of the latest version.

2. Implementing Upcasting with a Framework

Frameworks can provide a structured way to manage versioning and upcasting. The eventsourcing library for Python, which we will explore more in Module 7, has a particularly clear and conventional approach to this problem. Let's see how it works.

As you saw in the documentation, the pattern is straightforward:

• You add a class_version attribute to your event class.
• For each new version, you increment this number.
• You define a static method named upcast_vX_vY(state: dict), where X is the old version and Y is the new version. This method takes the dictionary of stored event attributes and modifies it to conform to the new schema.

Let's apply this to our recurring financial example.

Example: Evolving an FXTradeBooked Event

Imagine we have an FXTradeBooked event.

Version 1: The initial event captures the core details of the trade.

# V1 of our event
class FXTradeBooked(Aggregate.Created):
    ccy_pair: str
    rate: float
    amount: float
    client_id: str
    # class_version is implicitly 1

Sometime later, the business requires that all trades must have an explicit value_date (the settlement date). For old trades where this wasn't captured, the rule is to default to two business days after the trade was recorded.

Version 2: We introduce the new field and the upcaster.

# V2 of our event
class FXTradeBooked(Aggregate.Created):
    ccy_pair: str
    rate: float
    amount: float
    client_id: str
    value_date: str  # e.g., 'YYYY-MM-DD'

    class_version = 2

    @staticmethod
    def upcast_v1_v2(state: dict):
        # The 'state' dict contains the stored attributes of the V1 event.
        # We need to add the 'value_date'.
        # For this example, we'll use a placeholder logic.
        # In a real system, you might use the event's timestamp.
        state["value_date"] = "T+2" # Placeholder for calculated value date

When the eventsourcing library's repository loads an aggregate, it fetches the raw event data. If it finds an FXTradeBooked event stored with version 1, it will automatically call FXTradeBooked.upcast_v1_v2(stored_data) before using that data to construct the FXTradeBooked object. The aggregate's _apply method will then receive an object that reliably has the value_date attribute.

3. A Manual Implementation

While a library is convenient, understanding the underlying mechanism is crucial. A manual implementation makes the process explicit. Let's look at a simplified, framework-free approach.

Here, the responsibility for upcasting might fall to a dedicated EventUpcaster component, which is called by the repository before it deserializes the event data into a concrete object.

Building on that idea, let's create a more complete, runnable example that simulates the whole process: storing an old event, reading it, upcasting it, and applying it.

Full Python Example

This script simulates an event store and the rehydration process, showing the upcaster in action.

import uuid
from typing import Any, Dict

# --- Event Class Definitions ---

# Let's use simple dataclasses to represent our events
from dataclasses import dataclass

@dataclass
class FXTradeBooked:
    # This is the V2 schema, our "current" version
    trade_id: uuid.UUID
    ccy_pair: str
    rate: float
    amount: float
    client_id: str
    value_date: str  # The new field

# --- Upcasting Logic ---

def upcast(event_name: str, data: Dict[str, Any], version: int) -> Dict[str, Any]:
    """
    Applies upcasters to data until it reaches the current version.
    """
    if event_name == "FXTradeBooked":
        if version == 1:
            # Apply v1 -> v2 transformation
            data = upcast_fx_trade_booked_v1_to_v2(data)
            version = 2  # Update version after upcasting
        # Add other upcasters here if version > 2
        # if version == 2:
        #     data = upcast_fx_trade_booked_v2_to_v3(data)
        #     version = 3
    return data

def upcast_fx_trade_booked_v1_to_v2(state: Dict[str, Any]) -> Dict[str, Any]:
    """
    Transforms FXTradeBooked from version 1 to version 2.
    Adds a default 'value_date'.
    """
    print("--- Running upcaster: FXTradeBooked v1 -> v2 ---")
    state["value_date"] = "T+2" # Default value for old trades
    return state

# --- Simulation of an Event Store and Rehydration ---

# 1. Our "database" stores events as dictionaries with metadata
event_store_db = [
    {
        "event_name": "FXTradeBooked",
        "version": 1,
        "data": {
            "trade_id": uuid.uuid4(),
            "ccy_pair": "EUR/USD",
            "rate": 1.08,
            "amount": 100000,
            "client_id": "client-abc"
            # Note: 'value_date' is missing
        }
    }
]

# 2. The repository fetches the raw data and rehydrates the event object
def get_events_from_store():
    events = []
    for stored_event in event_store_db:
        # Fetch raw data
        event_name = stored_event["event_name"]
        version = stored_event["version"]
        data = stored_event["data"]

        # UPCASTING STEP: Transform data to the current version
        upcasted_data = upcast(event_name, data, version)

        # Deserialize into the current event object
        # In a real system, you'd map event_name to a class
        if event_name == "FXTradeBooked":
            event_obj = FXTradeBooked(**upcasted_data)
            events.append(event_obj)
    return events

# 3. Let's run the simulation
print("Rehydrating events from the store...")
rehydrated_events = get_events_from_store()

# 4. Verify the result
trade_event = rehydrated_events[0]
print("\nRehydrated Event Object:")
print(trade_event)

# The aggregate would now apply this fully-formed V2 event.
# Notice the 'value_date' is present, thanks to the upcaster.
assert hasattr(trade_event, 'value_date')
assert trade_event.value_date == "T+2"
print("\nAssertion successful: 'value_date' was added by the upcaster.")

Running this script would produce the following output:

Rehydrating events from the store...
--- Running upcaster: FXTradeBooked v1 -> v2 ---

Rehydrated Event Object:
FXTradeBooked(trade_id=UUID('...'), ccy_pair='EUR/USD', rate=1.08, amount=100000, client_id='client-abc', value_date='T+2')

Assertion successful: 'value_date' was added by the upcaster.

This example clearly separates the concerns:

• The FXTradeBooked class only knows about its current (V2) structure.
• The upcast_... function contains the specific, isolated transformation logic.
• The repository (get_events_from_store in our simulation) orchestrates the process of reading, upcasting, and deserializing.

Conclusion

Today we bridged the gap between the theory of event evolution and its practical implementation. You have now seen how to write and apply an upcaster function, a critical skill for maintaining event-sourced systems over the long term.

Key Takeaways:

• Upcasting is a run-time transformation that converts historical event data to the current schema before it reaches the domain model.
• The implementation requires two key pieces: a way to version events (class_version) and the transformation functions themselves (upcast_vX_vY).
• This can be managed by convention within a framework (like eventsourcing) or implemented manually with a dedicated upcasting component.
• The primary benefit is that your core business logic (in aggregates) remains clean and only ever deals with the latest, consistent version of events, drastically improving maintainability.

Preview of the Next Lesson:
We have now covered the core mechanics of event sourcing, from aggregates and event streams to snapshots and schema evolution. In our final lesson of this module, we will take a step back to evaluate the pattern as a whole. We will analyze the benefits and drawbacks of event sourcing, identifying use cases where it is and is not appropriate. This will equip you to make informed architectural decisions about when to employ this powerful but complex pattern.