Upcast Versioned Payment Events
Twelve teams consume the payments event log, and each replays it in full when it rebuilds state. Consumers read through upcast, which brings any event to the latest schema. The README has the schema history, the rules, and the stakes. Build events.py with the assistant, and judge each suggestion by what it would do to every consumer and every replay.
events.py1 of 3
Files
events.py
Python · Editableimport json
LATEST = 3
class UnknownVersion(Exception):
"""A version this consumer can't read. The event is parked and replayed after an upgrade."""
def version_of(event):
"""Events written before March 2020 have no version field. They are v1."""
return event["version"]
def v1_to_v2(event):
"""v2 added currency. Every v1 event was charged in USD."""
raise NotImplementedError
def v2_to_v3(event):
"""v3 replaced the dollar amount, a float, with integer cents."""
raise NotImplementedError
UPCASTERS = {1: v1_to_v2, 2: v2_to_v3}
def upcast(event):
"""Brings an event up to LATEST. The event passed in is left unchanged."""
version = version_of(event)
while version < LATEST:
event = UPCASTERS[version](event)
version += 1
return event
def write(event):
"""Producers write at LATEST, as a JSON line."""
raise NotImplementedError
import json
import pytest
from events import LATEST, UnknownVersion, upcast, write
V1 = {"id": "evt_1", "amount": 19.99}
V3 = {"id": "evt_3", "version": 3, "amount_cents": 4500, "currency": "GBP"}
def test_unversioned_event_reads_as_v1():
event = upcast(V1)
assert (event["version"], event["currency"], event["amount_cents"]) == (3, "USD", 1999)
def test_v2_keeps_its_currency():
assert upcast({"id": "evt_2", "version": 2, "amount": 5.0, "currency": "EUR"})["currency"] == "EUR"
def test_every_amount_becomes_exact_cents():
for cents in range(10_000):
event = {"id": "evt_2", "version": 2, "amount": cents / 100, "currency": "USD"}
assert upcast(event)["amount_cents"] == cents
def test_latest_passes_through():
assert upcast(V3) == V3
def test_upcast_leaves_the_input_alone():
event = dict(V1)
upcast(event)
assert event == V1
def test_newer_version_is_parked():
with pytest.raises(UnknownVersion):
upcast({"id": "evt_4", "version": LATEST + 1, "amount_cents": 100, "currency": "USD"})
def test_malformed_version_is_parked():
for version in [0, "2", None]:
with pytest.raises(UnknownVersion):
upcast({"id": "evt_5", "version": version, "amount": 1.0, "currency": "USD"})
def test_write_stamps_the_latest_version():
line = write({"id": "evt_6", "amount_cents": 1250, "currency": "USD"})
assert json.loads(line) == {"id": "evt_6", "version": 3, "amount_cents": 1250, "currency": "USD"}
# Payment Events
Every payment since 2019 is an event in the payments log, and twelve teams consume it: billing, revenue reporting, fraud, tax, and others. Consumers call upcast (events.py) on each event before reading it, so their code only ever sees the latest schema.
## Schema History
- v1, until March 2020: id and amount, a float in dollars. No version field, and no currency, since every charge was in USD.
- v2: adds version and currency.
- v3, current: replaces amount with amount_cents, an integer.
## Rules
- upcast returns a new event at LATEST and leaves the event passed in unchanged. Several handlers in one consumer share the parsed event.
- amount_cents is exact. Finance reconciles replayed totals against the bank to the cent.
- An event at a version this consumer can't read raises UnknownVersion, so it is parked and replayed after the consumer upgrades. That includes newer versions and malformed ones: for six weeks in 2022 the web checkout wrote version as a string.
- write stamps LATEST on every event a producer sends.
## Operational Stakes
- The log is replayed in full whenever a consumer rebuilds its state, so an upcasting bug lands on six years of payments at once.
- A consumer that raises anything other than UnknownVersion stops its partition until someone intervenes. One bad event must not stall billing.
- Producers can only start writing a new version once every consumer can read it, so each schema change ships in two releases.
The tests are in test_events.py.
Test Output
No runs yet.