Implement Exactly-Once Transaction Log in Python

A mock transaction log that deduplicates transaction IDs so each is recorded only once, with a dataclass for records and simple in-memory storage.

Easy Python 3.9+ Aug 9, 2026 Data pipelines & processing 14 views 0 copies

Python code

33 lines
Python 3.9+
from dataclasses import dataclass
from typing import Dict, Optional


@dataclass
class TxnRecord:
    txn_id: str
    status: str


class ExactlyOnceTxnLog:
    def __init__(self) -> None:
        self._log: Dict[str, TxnRecord] = {}
        self._processed_ids: set = set()

    def record(self, txn_id: str, status: str) -> bool:
        """Record a transaction exactly once. Returns True if newly recorded."""
        if txn_id in self._processed_ids:
            return False
        self._processed_ids.add(txn_id)
        self._log[txn_id] = TxnRecord(txn_id=txn_id, status=status)
        return True

    def get(self, txn_id: str) -> Optional[TxnRecord]:
        return self._log.get(txn_id)


if __name__ == "__main__":
    log = ExactlyOnceTxnLog()
    print(log.record("txn-001", "committed"))
    print(log.record("txn-001", "duplicate-attempt"))
    print(log.record("txn-002", "failed"))
    print(log.get("txn-001"))

Output

stdout
True
False
True
TxnRecord(txn_id='txn-001', status='committed')

How it works

The ExactlyOnceTxnLog class uses a set _processed_ids to track transaction IDs already seen, ensuring each transaction is recorded exactly once. The record method returns True only for new transactions; duplicates return False without updating the log. The _log dictionary stores the actual TxnRecord objects, and the get method safely retrieves a record or returns None if not found. The dataclass TxnRecord provides a lightweight, immutable-like structure for storing transaction status.

Common mistakes

  • Using a list to check duplicates, which is O(n) instead of O(1)
  • Forgetting to check for duplicates before writing, causing data loss or overwrites
  • Not returning a boolean from record(), making it hard to know if a transaction was actually applied

Variations

  1. Use a dictionary with the transaction ID as key and a tuple for status and a timestamp
  2. Persist the log to a database with a unique constraint on transaction ID

Real-world use cases

  • In a financial microservice, ensuring each payment request is processed only once despite retries.
  • In an event-driven pipeline, deduplicating incoming messages from a queue to avoid duplicate processing.
  • In an order processing system, logging each order idempotently to handle consumer replay and network retries.

Sponsored

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from Data pipelines & processing

Related tutorials and quizzes for this topic.