How to mock batch commit of transactions in Python
Simulate a transaction batch writer with commit, rollback, and summary logic to test database write patterns without a real database.
Python code
51 linesimport json
from datetime import datetime, timezone
class TransactionBatch:
def __init__(self):
self.pending = []
self.committed = []
self._log = []
def add(self, operation):
self.pending.append(operation)
def commit(self):
if not self.pending:
return False
batch_id = f"batch_{len(self.committed) + 1}_{int(datetime.now(timezone.utc).timestamp())}"
self.committed.append({
"id": batch_id,
"operations": self.pending[:],
"committed_at": datetime.now(timezone.utc).isoformat()
})
self._log.append(f"Committed {len(self.pending)} ops as {batch_id}")
self.pending = []
return True
def rollback(self):
rolled_back = self.pending[:]
self.pending = []
self._log.append(f"Rolled back {len(rolled_back)} ops")
return rolled_back
def summary(self):
return {
"pending": len(self.pending),
"committed_batches": len(self.committed),
"total_committed_ops": sum(len(b["operations"]) for b in self.committed)
}
if __name__ == "__main__":
batch = TransactionBatch()
batch.add({"type": "INSERT", "table": "users", "row": {"id": 1, "name": "Alice"}})
batch.add({"type": "UPDATE", "table": "users", "row": {"id": 1, "name": "Alice A."}})
batch.add({"type": "INSERT", "table": "orders", "row": {"id": 100, "user_id": 1}})
print("Before commit:", batch.summary())
ok = batch.commit()
print("Commit success:", ok)
print("After commit:", batch.summary())
print("First committed batch id:", batch.committed[0]["id"])
print("Log:", batch._log)
Output
Before commit: {'pending': 3, 'committed_batches': 0, 'total_committed_ops': 0}
Commit success: True
After commit: {'pending': 0, 'committed_batches': 1, 'total_committed_ops': 3}
First committed batch id: batch_1_1700000000
Log: ['Committed 3 ops as batch_1_1700000000']
How it works
The TransactionBatch class collects operations in a pending list and only flushes them to committed on commit(), which mimics a real database transaction. commit() generates a unique batch ID using a timestamp so you can trace which writes went together. rollback() clears pending operations and returns them for inspection, letting you simulate a failed transaction. The summary() method provides a quick snapshot of pending and committed counts, useful for verifying batch sizes in tests. The log keeps a lightweight audit trail for debugging and assertions.
Common mistakes
- Forgetting to clear `pending` after commit, causing duplicate writes
- Using a non-timezone-aware timestamp and getting inconsistent logs
- Modifying operations after commit because you didn't copy the list with `[:]`
- Assuming `commit()` always succeeds when `pending` is empty
Variations
- Use a `deque` for pending operations if you need fast pops from both ends
- Add a `flush()` method that combines commit and returns the batch ID
Real-world use cases
- Unit-testing a data pipeline that aggregates many inserts before writing to a database.
- Simulating a saga pattern where a batch of microservice calls commits or rolls back together.
- Prototyping a write-ahead log or event sourcing buffer before connecting a real database.
Sponsored
More from Database scaling & optimization
- Approximate Count with HyperLogLog in Python medium
- B-Tree Insert and In-Order Traversal in Python hard
- Broadcast a Small Reference Table in Python easy
- Build a Full Text Search Index in Python medium
- Build a Partial Index Mock in Python for Database Filtering easy
- Composite index leftmost prefix in Python medium
Keep learning
Related tutorials and quizzes for this topic.