"""Lab 01 — The integration fabric: ISO 20022, outbox, log, registry, products. An AI platform in a bank is only as useful as the systems it can reach, and those systems are thirty years old, extremely reliable, and were not designed with you in mind. Work top to bottom; each section's tests pass once that section is done: 1. ISO 20022 message identifiers 2. amounts and minor units 3. identifier checksums (IBAN, BIC) 4. the pain.001 parser 5. rails and finality 6. dual write, and the transactional outbox 7. a partitioned log 8. consumer groups and offsets 9. the idempotent consumer 10. schema compatibility and the registry 11. data products 12. reconciliation pytest # against your work LAB_MODULE=solution pytest # against the reference Determinism rules: the clock is injected, money is an integer number of MINOR units computed with Decimal (never float), partitioning is a derived digest (never ``hash()``), and every collection you return is sorted. """ from __future__ import annotations import hashlib import re import xml.etree.ElementTree as ET from dataclasses import dataclass, field, replace from decimal import Decimal, InvalidOperation from enum import Enum from typing import (Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple) # ====================================================================================== # 1. ISO 20022 — reading a message name # ====================================================================================== @dataclass(frozen=True) class MessageIdentifier: """`pain.001.001.09` — and every part of it means something. pain . 001 . 001 . 09 │ │ │ └── version │ │ └────────── variant (almost always 001) │ └────────────────── message number within the family └─────────────────────────── business area Knowing how to *read* the name is most of what a working knowledge of ISO 20022 means in an interview. The details are lookups; the families are the map. """ business_area: str message_number: str variant: str version: str @classmethod def parse(cls, name: str) -> "MessageIdentifier": """TODO: strict — four lowercase letters, then 3/3/2 digits. Raise ``ValueError`` on anything else, including an upper-case business area.""" raise NotImplementedError def __str__(self) -> str: raise NotImplementedError @property def family(self) -> str: raise NotImplementedError @property def description(self) -> str: """TODO: look up ``area.number``; ``"unknown message"`` when absent.""" raise NotImplementedError BUSINESS_AREAS: Mapping[str, str] = { "pain": "Payments Initiation", # customer -> bank "pacs": "Payments Clearing and Settlement", # bank -> bank "camt": "Cash Management", # statements, balances, investigations "acmt": "Account Management", "auth": "Authorities (regulatory reporting)", "reda": "Reference Data", "seev": "Securities Events", "setr": "Securities Trade", "tsmt": "Trade Services Management", } MESSAGE_CATALOGUE: Mapping[str, str] = { "pain.001": "CustomerCreditTransferInitiation", "pain.002": "CustomerPaymentStatusReport", "pain.008": "CustomerDirectDebitInitiation", "pacs.002": "FIToFIPaymentStatusReport", "pacs.004": "PaymentReturn", "pacs.008": "FIToFICustomerCreditTransfer", "pacs.009": "FinancialInstitutionCreditTransfer", "camt.052": "BankToCustomerAccountReport", "camt.053": "BankToCustomerStatement", "camt.054": "BankToCustomerDebitCreditNotification", "camt.056": "FIToFIPaymentCancellationRequest", } # ====================================================================================== # 2. pain.001 — parsing and validating # ====================================================================================== NS = "urn:iso:std:iso:20022:tech:xsd:pain.001.001.09" class Severity(str, Enum): ERROR = "error" # the message is rejected WARNING = "warning" # accepted, but somebody should look @dataclass(frozen=True) class Rejection: """A rejection names the ELEMENT, not just the problem. "Invalid message" sends a human to read 400 lines of XML. `[1]/` sends them to one line. In a bank the difference is a day. """ code: str element: str reason: str severity: Severity = Severity.ERROR def __str__(self) -> str: return f"[{self.code}] {self.element}: {self.reason}" @dataclass(frozen=True) class Party: name: str iban: Optional[str] = None bic: Optional[str] = None country: Optional[str] = None address_lines: Tuple[str, ...] = () @dataclass(frozen=True) class CreditTransfer: end_to_end_id: str amount_minor: int # integer minor units. Never a float. currency: str creditor: Party remittance_info: str = "" @dataclass(frozen=True) class PaymentInstruction: message_id: str creation_datetime: str number_of_transactions: int control_sum_minor: int initiating_party: str payment_information_id: str requested_execution_date: str debtor: Party transfers: Tuple[CreditTransfer, ...] @property def actual_sum_minor(self) -> int: raise NotImplementedError #: Currencies whose minor unit is not 2 digits. Assuming 2 everywhere is the classic #: cross-border bug: JPY 1000 parsed as 100000 minor units is a payment 100x too large. CURRENCY_EXPONENT: Mapping[str, int] = { "JPY": 0, "KRW": 0, "VND": 0, "CLP": 0, "ISK": 0, "BHD": 3, "KWD": 3, "OMR": 3, "TND": 3, "JOD": 3, } DEFAULT_EXPONENT = 2 VALID_CURRENCIES = frozenset({ "AED", "USD", "EUR", "GBP", "JPY", "SAR", "KWD", "BHD", "CHF", "SGD", "INR"}) def currency_exponent(currency: str) -> int: """TODO: the minor-unit exponent, case-insensitively; 2 by default.""" raise NotImplementedError def to_minor(amount: str, currency: str) -> int: """TODO: parse a decimal amount string into integer minor units. Use ``Decimal``, never ``float``: ``1.15 * 100`` is ``114.99999999999999``, and ``int()`` of that is one cent short — silently, on every payment. Raise ``ValueError`` for a non-numeric amount, and for one with MORE decimal places than the currency permits. Do not round: silently rounding a payment amount is how a reconciliation break happens every day. """ raise NotImplementedError def from_minor(minor: int, currency: str) -> str: """TODO: the inverse. ``from_minor(to_minor(a, c), c) == a`` for well-formed a.""" raise NotImplementedError def valid_iban(candidate: str) -> bool: """TODO: ISO 13616 mod-97 — the same check as Phase 11. Banks reuse their checksums. Strip spaces, upper-case, require 2 letters + 2 digits + 11-30 alphanumerics. Move the first four characters to the end, map letters to ``ord(c) - 55``, and require ``int(...) % 97 == 1``. """ raise NotImplementedError def valid_bic(candidate: str) -> bool: """TODO: ISO 9362 — 4 LETTERS (institution) + 2 letters (country) + 2 alphanumerics (location), optionally + 3 more (branch). Note the institution code is letters only; accepting digits there is the common mistake.""" raise NotImplementedError class Pain001Parser: """Parse and validate a pain.001 CustomerCreditTransferInitiation. The design rule: **collect every rejection, then decide.** A parser that raises on the first problem forces the sender into one round trip per error, and a payment file with six problems takes six days to fix. """ def __init__(self, *, max_transactions: int = 1000) -> None: self.max_transactions = max_transactions def parse(self, xml: str) -> Tuple[Optional[PaymentInstruction], List[Rejection]]: """TODO: return ``(instruction_or_None, rejections_sorted_by_(element, code))``. The codes the tests pin: | code | meaning | |---|---| | FF01 | malformed XML | | FF02 | wrong namespace — a version mismatch is not a warning | | FF03 | ReqdExctnDt is not an ISO date | | FF04 | unsupported currency | | FF05 | unparseable amount | | FF07 | IBAN fails mod-97 | | FF08 | invalid BIC | | MS01 | a mandatory element is absent | | MS02 | the Ccy attribute is missing from an amount | | AM01 | amount is not strictly positive | | DU01 | duplicate EndToEndId **within one file** | | CS01 | NbOfTxs disagrees with the transaction count | | CS02 | CtrlSum disagrees with the sum of the amounts | | LM01 | more transactions than the configured limit | Three things worth getting right: * **the namespace pins the version** — a .001.09 parser must not silently accept a .001.03 document whose element semantics differ; * **CS01/CS02 are cross-field checks** a per-element schema cannot express, and they are what catch a truncated file; * **DU01 matters** because downstream idempotency keys on EndToEndId, so a duplicate inside one file is how the same payment gets made twice. Return ``None`` for the instruction whenever any ERROR-severity rejection exists. """ raise NotImplementedError # ====================================================================================== # 3. Payment rails and finality # ====================================================================================== class Finality(str, Enum): """When a payment stops being reversible — a SCHEME rule, not a database property. This is the phase's link back to Phase 10's side-effect classes: `irreversible` is not an engineering opinion about difficulty, it is a statement about what the scheme permits. An agent's autonomy is bounded by the rail it is using. """ REVOCABLE = "revocable" # before cut-off: cancel freely CONDITIONALLY_REVOCABLE = "conditionally_revocable" # a recall; may be refused FINAL = "final" # settled; only a NEW payment moves it back @dataclass(frozen=True) class Rail: rail_id: str name: str currency: str cutoff_minute: int # minutes past midnight, local settles_after_minutes: int max_amount_minor: Optional[int] = None supports_recall: bool = True def finality_at(self, submitted_minute: int, now_minute: int) -> Finality: """TODO: **settlement dominates the cut-off** — check it FIRST. 1. settled (``now >= submitted + settles_after``) -> FINAL; 2. else both before the cut-off -> REVOCABLE; 3. else CONDITIONALLY_REVOCABLE, or FINAL when the rail has no recall. The ordering is easy to get backwards: an *instant* payment settles in zero minutes, so it is FINAL the moment it is submitted — despite having no cut-off at all. Checking the cut-off first would report the most irrevocable rail in the bank as revocable. """ raise NotImplementedError RAILS: Mapping[str, Rail] = { "instant": Rail("instant", "Instant Payments (IPI)", "AED", cutoff_minute=1440, settles_after_minutes=0, max_amount_minor=50_000_00, supports_recall=False), "rtgs": Rail("rtgs", "UAEFTS (RTGS)", "AED", cutoff_minute=15 * 60, settles_after_minutes=15, supports_recall=False), "ach": Rail("ach", "WPS/ACH batch", "AED", cutoff_minute=14 * 60, settles_after_minutes=24 * 60, supports_recall=True), "swift": Rail("swift", "SWIFT cross-border", "USD", cutoff_minute=16 * 60, settles_after_minutes=48 * 60, supports_recall=True), } def choose_rail(amount_minor: int, currency: str, now_minute: int, *, urgent: bool = False) -> Tuple[Optional[str], str]: """TODO: pick the FASTEST eligible rail, and say WHY — including why the others were excluded. Return ``(rail_id_or_None, reason)``; the reason is never empty. Exclusions, in order: wrong currency, above the rail's limit, past its cut-off, and (when urgent) settling in more than an hour. An agent proposing a payment must be able to explain the rail choice, because the rail determines the finality, which determines whether a human must approve. """ raise NotImplementedError # ====================================================================================== # 4. The transactional outbox # ====================================================================================== @dataclass(frozen=True) class OutboxRecord: seq: int aggregate_id: str event_type: str payload: Mapping[str, Any] created_at: int published_at: Optional[int] = None @property def published(self) -> bool: raise NotImplementedError class DualWriteStore: """A deliberately BROKEN store, to demonstrate the problem the outbox solves. The dual-write problem: writing to the database and publishing to a broker are two separate systems, so a crash between them leaves them disagreeing — and there is no ordering of the two writes that fixes it. * DB then publish -> the crash LOSES the event. The payment happened; nothing downstream knows. * Publish then DB -> the crash produces a PHANTOM event. Downstream reacts to a payment that does not exist. The second is worse, and neither is acceptable. """ def __init__(self, *, broker: "MessageLog", fail_after_db: bool = False) -> None: raise NotImplementedError def save_and_publish(self, key: str, row: Mapping[str, Any], topic: str) -> None: """TODO: write the row, then (unless ``fail_after_db``) publish. Raise in between when it is set — that raise IS the lesson.""" raise NotImplementedError class Outbox: """Domain change and outbound event in ONE local transaction; relay afterwards. The insight is that both writes go to the *same* database, so the database's own atomicity covers them. A separate relay then reads unpublished rows and publishes them — at-least-once, because the relay can crash after publishing and before marking. Which is exactly why the consumer must be idempotent (§6). The outbox does not give you exactly-once delivery; it gives you *no lost events*, and idempotency turns that into exactly-once effects. """ def __init__(self, *, now: Callable[[], int]) -> None: raise NotImplementedError def transact(self, key: str, row: Mapping[str, Any], *, event_type: str, payload: Optional[Mapping[str, Any]] = None, fail_before_commit: bool = False) -> OutboxRecord: """TODO: both writes, atomically. ``fail_before_commit`` must leave **neither** — and must not consume a sequence number, because a gap in the sequence is indistinguishable from a lost record. The payload defaults to the row. """ raise NotImplementedError def unpublished(self) -> List[OutboxRecord]: raise NotImplementedError def mark_published(self, seq: int) -> None: raise NotImplementedError def records(self) -> List[OutboxRecord]: raise NotImplementedError class OutboxRelay: """Publishes unpublished outbox rows, **in sequence order**. ``fail_after_publish`` models the crash that makes at-least-once real: the event reached the broker, the row was not marked, and the next run publishes it again. """ def __init__(self, *, outbox: Outbox, log: "MessageLog", topic: str) -> None: raise NotImplementedError def run_once(self, *, fail_after_publish: bool = False) -> int: """TODO: publish each unpublished record in seq order, marking as you go, and return how many you published. With ``fail_after_publish``, publish exactly one and return WITHOUT marking it.""" raise NotImplementedError # ====================================================================================== # 5. A partitioned log # ====================================================================================== @dataclass(frozen=True) class Message: partition: int offset: int key: str value: Mapping[str, Any] timestamp: int class MessageLog: """A partitioned, append-only log — Kafka and Event Hubs are the same model. | Kafka | Event Hubs | |---|---| | topic | event hub | | partition | partition | | consumer group | consumer group | | offset | offset / sequence number | **Ordering is per partition, never across.** That is the single most important property, and the routing decision below is what makes it useful: keying by account means every event for one account lands in one partition, so per-account order holds even though global order does not. """ def __init__(self, *, partitions: int = 4, now: Optional[Callable[[], int]] = None) -> None: # TODO: raise ValueError below 1 partition. raise NotImplementedError def partition_for(self, key: str) -> int: """TODO: a DERIVED digest (``hashlib.blake2b``) mod the partition count. Never ``hash()``: Python salts string hashing per process, so the same key would route to different partitions across restarts and silently break ordering — the one guarantee the whole model provides. """ raise NotImplementedError def append(self, topic: str, key: str, value: Mapping[str, Any]) -> Message: """TODO: route by key, offset = the partition's current length.""" raise NotImplementedError def read(self, topic: str, partition: int, *, from_offset: int = 0, limit: Optional[int] = None) -> List[Message]: raise NotImplementedError def all(self, topic: str) -> List[Message]: """TODO: every message, partition then offset. NOT a global ordering — it is a deterministic *listing*, and the distinction matters.""" raise NotImplementedError def high_water_mark(self, topic: str, partition: int) -> int: raise NotImplementedError class ConsumerGroup: """Offsets per (topic, partition), committed explicitly. Two ordering rules that follow directly from the model: * **commit AFTER processing** — commit first and a crash loses the message (at-most-once); * **at-least-once is the practical default** — so processing must be idempotent. """ def __init__(self, group_id: str, log: MessageLog) -> None: raise NotImplementedError def committed(self, topic: str, partition: int) -> int: raise NotImplementedError def poll(self, topic: str, *, max_records: int = 10) -> List[Message]: """TODO: from each partition's committed offset, up to ``max_records`` total.""" raise NotImplementedError def commit(self, topic: str, partition: int, offset: int) -> None: """TODO: the committed offset is the NEXT one to read, not the last one read. Off-by-one here means either re-reading the last message forever or skipping it, and which one you get depends on a convention nobody wrote down. Raise ``ValueError`` on an attempt to rewind — a rewind must be deliberate. """ raise NotImplementedError def seek(self, topic: str, partition: int, offset: int) -> None: """TODO: deliberate replay, clamped at 0. Separate from ``commit`` so a rewind is never accidental.""" raise NotImplementedError def lag(self, topic: str) -> Dict[int, int]: raise NotImplementedError def total_lag(self, topic: str) -> int: raise NotImplementedError # ====================================================================================== # 6. The idempotent consumer # ====================================================================================== class IdempotentConsumer: """At-least-once delivery + idempotent handling = **exactly-once effects**. Exactly-once *delivery* is impossible (two generals). Exactly-once *effects* is routine, and it is what anybody actually wants. The whole mechanism is: derive a stable id from the message, remember the ids you have processed, skip the repeats. """ def __init__(self, group: ConsumerGroup, handler: Callable[[Message], Any], *, key_of: Optional[Callable[[Message], str]] = None) -> None: """TODO: default ``key_of`` derives from the message's CONTENT — never from the offset, because a replay under a different partition assignment yields different offsets for the same event.""" raise NotImplementedError def consume(self, topic: str, *, max_records: int = 100, fail_before_commit: bool = False) -> int: """TODO: poll, skip already-processed ids (counting them as ``duplicates``), handle the rest, and commit ``offset + 1`` after each. ``fail_before_commit`` returns after the FIRST message's effect without committing — so the next poll re-delivers it, and the dedup set is what makes that harmless. That is the whole argument, demonstrated. """ raise NotImplementedError # ====================================================================================== # 7. The schema registry # ====================================================================================== class Compatibility(str, Enum): """The mode IS the deploy order. This is the whole content of the section. | Mode | New schema readable by | Upgrade first | |---|---|---| | BACKWARD | the NEW reader, on OLD data | **consumers** | | FORWARD | the OLD reader, on NEW data | **producers** | | FULL | both | either | | NONE | nothing guaranteed | pray | """ BACKWARD = "backward" FORWARD = "forward" FULL = "full" NONE = "none" @dataclass(frozen=True) class Field: name: str type: str required: bool = True default: Any = None @dataclass(frozen=True) class Schema: subject: str version: int fields: Tuple[Field, ...] def field(self, name: str) -> Optional[Field]: raise NotImplementedError def check_backward(old: Schema, new: Schema) -> List[str]: """TODO: can a NEW reader read OLD data? Consumers upgrade first. Return problems. * a new REQUIRED field with no default breaks it — old data lacks it and there is nothing to fall back on. (Which is why "always give new fields a default" is such durable advice.) * a type change breaks it. Removing a field is fine: the new reader simply does not look for it. """ raise NotImplementedError def check_forward(old: Schema, new: Schema) -> List[str]: """TODO: can an OLD reader read NEW data? Producers upgrade first. * removing a REQUIRED field breaks it — the old reader still demands it; * a type change breaks it. Adding a required field is fine: the old reader ignores what it does not know. """ raise NotImplementedError def check_compatibility(old: Schema, new: Schema, mode: Compatibility) -> List[str]: """TODO: dispatch. FULL is the union of both; NONE is always empty.""" raise NotImplementedError #: TODO: fill in. BACKWARD -> "consumers first, then producers"; FORWARD -> the reverse; #: FULL -> "either order"; NONE -> "coordinate manually; there is no guarantee". DEPLOY_ORDER: Mapping[Compatibility, str] = {} class SchemaRegistry: def __init__(self, *, default_mode: Compatibility = Compatibility.BACKWARD) -> None: raise NotImplementedError def set_mode(self, subject: str, mode: Compatibility) -> None: raise NotImplementedError def mode(self, subject: str) -> Compatibility: raise NotImplementedError def register(self, subject: str, fields: Sequence[Field]) -> Schema: """TODO: version 1 registers unconditionally; later versions must pass the subject's compatibility check or raise ``ValueError`` naming the mode. A REFUSED registration must not create a version — otherwise the version numbers lie about what was ever live. """ raise NotImplementedError def latest(self, subject: str) -> Schema: raise NotImplementedError def version(self, subject: str, version: int) -> Schema: raise NotImplementedError def deploy_order(self, subject: str) -> str: raise NotImplementedError # ====================================================================================== # 8. Data products # ====================================================================================== @dataclass(frozen=True) class QualityRule: name: str check: Callable[[Sequence[Mapping[str, Any]]], bool] description: str @dataclass(frozen=True) class DataProductContract: """Schema, semantics, quality, freshness, ownership. All five, or it is a table. The freshness SLO is the one that separates a data *product* from a dataset: it is a promise with a number, which means it can be breached, which means somebody can be told. """ product_id: str owner: str # a named human schema: Schema freshness_slo_ticks: int quality_rules: Tuple[QualityRule, ...] = () classification: str = "internal" description: str = "" @dataclass(frozen=True) class ContractResult: product_id: str passed: bool schema_problems: Tuple[str, ...] quality_failures: Tuple[str, ...] freshness_lag: int fresh: bool def report(self) -> str: """TODO: one readable line, naming the product and every failing dimension.""" raise NotImplementedError class DataProduct: def __init__(self, contract: DataProductContract, *, now: Callable[[], int]) -> None: raise NotImplementedError def publish(self, rows: Sequence[Mapping[str, Any]]) -> None: """TODO: replace the rows and reset the freshness clock.""" raise NotImplementedError def check(self) -> ContractResult: """TODO: all three dimensions. * **schema** — every required field present, every present field the right type. Note ``bool`` is a subclass of ``int``. Cap the reported problems at 5 and sort them, so a 50-row failure produces a readable line; * **quality** — every rule, naming the failures; * **freshness** — ``now() - last_updated <= freshness_slo_ticks``. """ raise NotImplementedError # ====================================================================================== # 9. Reconciliation # ====================================================================================== class BreakType(str, Enum): MISSING_IN_B = "missing_in_b" MISSING_IN_A = "missing_in_a" VALUE_MISMATCH = "value_mismatch" @dataclass(frozen=True) class Break: break_type: BreakType key: str a_value: Optional[int] b_value: Optional[int] def __str__(self) -> str: """TODO: ``"key: A=1 B=2"`` for a mismatch, ``"key: missing_in_b"`` otherwise.""" raise NotImplementedError def reconcile(a: Mapping[str, int], b: Mapping[str, int]) -> List[Break]: """TODO: compare two independent records; return every break, SORTED BY KEY. The bank habit worth adopting everywhere. A break is not an error to be swallowed — it is a *finding* with an owner and a due date. The value of reconciliation is not that it is clever; it is that it runs every day and somebody is accountable for the breaks. This is what would have caught every dual-write bug in this file. """ raise NotImplementedError # ====================================================================================== # Worked example # ====================================================================================== def main() -> None: # pragma: no cover """TODO (optional): once the tests pass, build the twelve-section demo. Compare against ``python solution.py`` — but only after your own runs. """ raise NotImplementedError if __name__ == "__main__": # pragma: no cover main()