Reference library

Database scaling & optimization

Indexing, connection pooling, read replicas, query tuning, and throughput-aware SQL.

3 matches

How to Mock a Cross-Shard Saga in Python

Simulate a distributed saga with compensating transactions across multiple database shards using a lightweight Python class that tracks executed steps and rolls them back in reverse on failure.

saga sharding distributed-systems

Python

import json


class SagaState:
    def __init__(self, saga_id):
        self.saga_id = saga_id
        self.executed_steps = []
        self.compensations = []

    def execute_step(self, shard, step_name, operation):
        self.executed_steps.append((shard, step_name))
        print(f"[Saga {self.saga_id}] Executin…

14 0 Open

Database scaling & optimization medium

How to Simulate Distributed Transactions in Python with a Mock

Model distributed transaction behavior with a mock Transaction class that supports commit, rollback, and failure simulation.

transactions mock database

Python

class Transaction:
    def __init__(self, id):
        self.id = id
        self.operations = []
        self.committed = False

    def add_operation(self, op, data):
        self.operations.append((op, data))

    def commit(self):
        if not self.operations:
            raise ValueError("No operations to commit…

13 0 Open

Database scaling & optimization medium

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.

transactions mock batch

Python

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 …

16 0 Open

Browse by section

Each section groups closely related Python snippets.