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.

Medium Python 3.9+ Aug 9, 2026 Database scaling & optimization 16 views 0 copies

Python code

51 lines
Python 3.9+
import 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

stdout
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

  1. Use a `deque` for pending operations if you need fast pops from both ends
  2. 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

Run this sample

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

Open editor

More from Database scaling & optimization

Related tutorials and quizzes for this topic.