How to Implement a Two-Phase Commit Mock in Python

import random
from dataclasses import dataclass
from typing import Dict, List, Optional


@dataclass
class Transaction:
    tx_id: int
    data: Dict[str, str]


class TwoPhaseCommitMock:
    """Simple two-phase commit mock with prepare and commit phases."""

    def __init__(self) -> None:
        self.prepared: List[int] = []
        self.committed: Dict[int, Transaction] = {}
        self.fail_on_prepare: set[int] = set()
        self.fail_on_commit: set[int] = set()

    def prepare(self, tx: Transaction) -> bool:
        # Simulate possible failure during prepare phase
        if tx.tx_id in self.fail_on_prepare or random.random() < 0.1:
            return False
        self.prepared.append(tx.tx_id)
        return True

    def commit(self, tx: Transaction) -> bool:
        if tx.tx_id not in self.prepared:
            print(f"Transaction {tx.tx_id} not prepared, cannot commit")
            return False
        if tx.tx_id in self.fail_on_commit:
            return False
        self.committed[tx.tx_id] = tx
        return True

    def abort(self, tx: Transaction) -> None:
        if tx.tx_id in self.prepared:
            self.prepared.remove(tx.tx_id)
            print(f"Transaction {tx.tx_id} aborted")

    def run_transaction(self, tx: Transaction) -> bool:
        """Execute a full two-phase commit transaction."""
        if not self.prepare(tx):
            print(f"Prepare failed for transaction {tx.tx_id}, aborting")
            self.abort(tx)
            return False
        if not self.commit(tx):
            print(f"Commit failed for transaction {tx.tx_id}, aborting")
            self.abort(tx)
            return False
        print(f"Transaction {tx.tx_id} committed successfully")
        return True


if __name__ == "__main__":
    # Deterministic mock for demonstration
    mock = TwoPhaseCommitMock()
    mock.fail_on_prepare = {2}  # Always fail on transaction 2
    mock.fail_on_commit = {3}   # Always fail on commit for transaction 3

    t1 = Transaction(1, {"account": "A", "amount": "100"})
    t2 = Transaction(2, {"account": "B", "amount": "50"})
    t3 = Transaction(3, {"account": "C", "amount": "75"})

    for tx in [t1, t2, t3]:
        mock.run_transaction(tx)

    print(f"\nCommitted transactions: {sorted(mock.committed.keys())}")
    print(f"Prepared (but maybe uncommitted): {mock.prepared}")