Reference library

Database scaling & optimization

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

21 matches
Database scaling & optimization medium

Approximate Count with HyperLogLog in Python

A mock HyperLogLog implementation uses hash-based registers to estimate cardinality of large datasets with sublinear memory.

hyperloglog cardinality hash
Python
import hashlib

class HyperLogLog:
    def __init__(self, precision=4):
        if precision < 4 or precision > 16:
            raise ValueError("precision must be between 4 and 16")
        self.precision = precision
        self.registers = [0] * (1 << precision)

    def _hash(self, value):
        return int(hashl…
15 0 Open
Database scaling & optimization medium

Build a Full Text Search Index in Python

Create a simple inverted index for full-text search with the standard library, supporting multi-word AND queries across documents.

search inverted-index text-processing
Python
import re
from collections import defaultdict


class SimpleTextIndex:
    def __init__(self):
        self.index = defaultdict(list)
        self.documents = {}

    def add_document(self, doc_id, text):
        self.documents[doc_id] = text
        words = set(re.findall(r'\w+', text.lower()))
        for word in wo…
12 0 Open
Database scaling & optimization medium

Composite index leftmost prefix in Python

Simulate a composite index in SQLite and check whether query columns match the leftmost prefix rule for index usage.

sqlite indexes database
Python
import sqlite3


def get_indexed_columns(table_name):
    """Simulate a composite index by reading column names that start with 'idx_'."""
    conn = sqlite3.connect(":memory:")
    conn.execute(f"CREATE TABLE {table_name} (id INTEGER, idx_col1 TEXT, idx_col2 INTEGER, other TEXT)")
    conn.execute(f"CREATE INDEX idx_…
14 0 Open
Database scaling & optimization medium

Consistent Hashing with Virtual Buckets in Python

This code maps many virtual buckets onto a few physical buckets using a consistent hashing ring, ensuring balanced distribution with minimal remapping when physical buckets change.

consistent-hashing virtual-buckets sharding
Python
import random

class VirtualBuckets:
    """Maps many virtual buckets onto few physical buckets using consistent hashing."""
    
    def __init__(self, physical_buckets, virtual_factor=100):
        self.physical = list(physical_buckets)
        self.virtual_factor = virtual_factor
        self.ring = []
        self…
13 0 Open
Database scaling & optimization medium

How to Build a Connection Pool Reuse Mock in Python

Build a mock connection pool with context manager to track connection reuse, acquires, and releases in Python.

connection-pool context-manager database
Python
import time
from contextlib import contextmanager


class Connection:
    def __init__(self, name):
        self.name = name
        self.in_use = False
        self.busy_since = None

    def fetch(self):
        return f"data from {self.name}"


class ConnectionPool:
    def __init__(self, size=3):
        self.conn…
13 0 Open
Database scaling & optimization medium

How to Build a Shard Map Mock Dict in Python

Implement a dictionary-like class that distributes keys across multiple shards using Python's hash() for realistic data partitioning.

dict sharding hash
Python
class ShardMap:
    def __init__(self, shard_count):
        self.shards = {i: {} for i in range(shard_count)}
        self.shard_count = shard_count

    def _shard_for(self, key):
        return hash(key) % self.shard_count

    def __getitem__(self, key):
        return self.shards[self._shard_for(key)][key]

    d…
14 0 Open
Database scaling & optimization medium

How to Create a Covering Index with INCLUDE Columns in Python

Create a covering index with INCLUDE columns in SQLite from Python and inspect the query plan to confirm the index covers the query.

sqlite indexing covering index
Python
import sqlite3

def create_covering_index_mock():
    conn = sqlite3.connect(":memory:")
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE employees (
            id INTEGER PRIMARY KEY,
            name TEXT,
            department TEXT,
            salary INTEGER
        )
    """)

    employe…
15 0 Open
Database scaling & optimization medium

How to Eager Load with JOIN to Reduce N+1 Queries in Python

Demonstrates eager loading with a SQL JOIN to reduce N+1 query patterns down to a single database call when fetching related data.

eager-loading n-plus-1 join
Python
import sqlite3


def eager_load_join_reduce(mock_db_path=":memory:"):
    """Demonstrate eager loading where joins reduce query count from N+1 to 1."""
    conn = sqlite3.connect(mock_db_path)
    cursor = conn.cursor()
    cursor.executescript(
        """
        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TE…
16 0 Open
Database scaling & optimization medium

How to Implement Consistent Hashing in Python

Build a consistent hash ring in Python that distributes keys across nodes and minimizes remapping when nodes are added or removed.

consistent-hashing distributed-systems sharding
Python
import hashlib
from bisect import bisect_right


class ConsistentHashRing:
    def __init__(self, nodes, replicas=3):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
        for node in nodes:
            self.add_node(node)

    def _hash(self, key):
        return int(hashlib.md…
15 0 Open
Database scaling & optimization medium

How to Implement Read-After-Write Consistency Mock in Python

Simulate strong versus eventual read-after-write consistency with a primary and replica store, demonstrating the difference in data visibility over time.

consistency replication mock
Python
import time


class MockStorage:
    def __init__(self, write_delay=0.1):
        self.store = {}
        self.replica = {}
        self.write_delay = write_delay

    def write(self, key, value):
        # Write to primary storage immediately
        self.store[key] = value
        # Simulate async replication delay
…
13 0 Open
Database scaling & optimization medium

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 Mock a Hot Shard Split in Python

Simulate a database hot shard splitting into two shards by key ranges when it exceeds a threshold, with a mock class for testing.

sharding databases mock
Python
import random
from collections import defaultdict


class HotShardMock:
    """Mock implementation of a hot shard split in a distributed database."""

    def __init__(self, shard_id="shard_1", max_entries=5):
        self.shard_id = shard_id
        self.max_entries = max_entries
        self.entries = {}

    def ad…
16 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 Simulate a Stable Sort Cursor in Python

Build a MongoDB-style cursor mock that stably sorts records by a key while preserving original order for ties, with next() and rewind() methods.

sorting cursors database
Python
```python
import random

class CursorStableSortMock:
    """Simulates stable sorting with a cursor-like pointer for MongoDB-style queries."""
    
    def __init__(self, data, sort_key, reverse=False):
        self.data = list(data)
        self.sort_key = sort_key
        self.reverse = reverse
        self._index = …
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
Database scaling & optimization medium

Idempotent Writes for Sharded Databases in Python

Implement a mock shard with idempotent write support using request IDs to prevent duplicate writes and track the latest value per key.

idempotency sharding distributed-systems
Python
import json


class ShardMock:
    """Mock distributed shard with idempotent write support."""

    def __init__(self, shard_id):
        self.shard_id = shard_id
        self._store = {}

    def write(self, key, value, request_id):
        """Write value only if request_id not yet processed; idempotent."""
        i…
13 0 Open
Database scaling & optimization medium

Mock CQRS Read/Write Split in Python

Separate order mutations from queries using a read model and write model to mock CQRS-style separation of concerns.

cqrs read-write dataclass
Python
from dataclasses import dataclass, field
from typing import List, Dict


@dataclass
class Order:
    id: int
    amount: float
    status: str = "pending"


class OrderWriteModel:
    """Handles all mutations (writes) to orders."""

    def __init__(self):
        self._orders: Dict[int, Order] = {}
        self._next…
12 0 Open
Database scaling & optimization medium

Offset vs Keyset Pagination in Python

Demonstrate offset-based pagination and keyset (cursor) pagination with a simple in-memory dataset, showing how each returns pages of records.

pagination keyset offset
Python
"""Demonstrate pagination using offset vs keyset (cursor) approach."""

ITEMS = [
    {"id": 1, "name": "Alice"},
    {"id": 2, "name": "Bob"},
    {"id": 3, "name": "Carol"},
    {"id": 4, "name": "David"},
    {"id": 5, "name": "Eve"},
]

def offset_paginate(items, page, page_size):
    """Return a page using offset…
14 0 Open
Database scaling & optimization medium

Simulate a GIN Index for JSONB in Python

Build a mock Generalized Inverted Index (GIN) that flattens JSON documents into key-value tokens for fast lookup queries, mimicking PostgreSQL JSONB indexing.

jsonb gin-index inverted-index
Python
import json
import random
from collections import defaultdict

# Mock GIN (Generalized Inverted Index) for JSONB key-value pairs
class GINIndex:
    def __init__(self):
        self.posting_lists = defaultdict(list)  # token -> list of doc_ids
    
    def index(self, doc_id, json_obj):
        """Index a JSON documen…
14 0 Open
Database scaling & optimization medium

Snowflake ID Generator with Cluster Index Mock in Python

A thread-safe Snowflake ID generator mock that creates unique 64-bit IDs across simulated cluster nodes and maintains a sorted in-memory index for range queries.

snowflake id-generation clustering
Python
import time
import threading

class SnowflakeIDGenerator:
    def __init__(self, machine_id, datacenter_id):
        self.machine_id = machine_id
        self.datacenter_id = datacenter_id
        self.sequence = 0
        self.last_timestamp = -1
        self.machine_bits = 5
        self.datacenter_bits = 5
        …
13 0 Open
Database scaling & optimization medium

Two Phase Commit Cross Shard Mock in Python

Simulates a two-phase commit across shards with failure handling to demonstrate distributed transaction coordination in Python.

two-phase-commit distributed-systems transaction
Python
"""Mock cross-shard two-phase commit with caution handling."""

class Shard:
    def __init__(self, name):
        self.name = name
        self.prepared = False
        self.committed = False
        self.aborted = False

    def prepare(self):
        # Simulate potential failure (1 in 3 chance on third shard)
     …
14 0 Open

Browse by section

Each section groups closely related Python snippets.

Database scaling & optimization — Python code examples

What you will find here

This page collects database scaling & optimization snippets — short, copy-ready Python you can paste into our free online IDE and run without installing anything. Each sample includes a plain-English explanation and the full source code.

Samples vs tutorials and challenges

Samples are quick reference — one concept per page. For step-by-step teaching, use our Python tutorials. To test yourself, try quizzes or coding challenges. Clean up style with the Python formatter.