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.
Python code
33 linesfrom 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
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
- Use a dictionary with the transaction ID as key and a tuple for status and a timestamp
- 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
More from Data pipelines & processing
- Add a UUID Surrogate Key to Each Row in a CSV with Python easy
- Attach Source File Metadata to Records in Python easy
- Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets medium
- Check Null Rate Threshold in PySpark DataFrame medium
- Count Records Processed per Category in Python easy
- Create Data Helper Functions in Python for Beginners easy
Keep learning
Related tutorials and quizzes for this topic.