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

Cross Shard Query Scatter Gather Mock in Python

Simulate a distributed database cross-shard query using a scatter-gather pattern with a mock Python implementation.

scatter-gather sharding distributed-systems
Python
from dataclasses import dataclass
from typing import List, Dict


@dataclass
class NodeResponse:
    node_id: int
    data: Dict[str, float]


def mock_query_shard(shard_id: int, shard_data: Dict[str, float], query: str) -> NodeResponse:
    """Simulate querying a single shard, returning matches whose value > 50."""
 …
13 0 Open
Database scaling & optimization medium

Database Helper in Python with SQLite Scaling Optimization

Build a beginner-friendly SQLite database helper class with WAL, indexed queries, and efficient batch inserts for scaling.

sqlite database scalability
Python
import sqlite3
from contextlib import contextmanager


class DatabaseHelper:
    """Beginner-friendly helper for SQLite database operations with scaling tips."""

    def __init__(self, db_path):
        self.db_path = db_path

    @contextmanager
    def connection(self):
        """Context manager for automatic comm…
14 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 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 Explain SQLite Query Plans in Python

Build a Python function that runs EXPLAIN QUERY PLAN on SQLite in-memory tables and prints the optimizer's execution plan for any SELECT statement.

sqlite query-plan optimization
Python
import sqlite3

def explain_query(sql: str) -> str:
    """Return the SQLite query plan for the given SQL statement."""
    conn = sqlite3.connect(":memory:")
    cursor = conn.cursor()
    
    # Create sample data for a realistic plan
    cursor.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
    c…
14 0 Open
Database scaling & optimization medium

How to Implement Keyset Pagination in Python (Seek Method)

Implement keyset (seek) pagination in Python with a mock paginator that efficiently fetches pages based on the last row rather than OFFSET.

pagination keyset seek-method
Python
from dataclasses import dataclass
from typing import List, Optional


@dataclass
class Row:
    id: int
    name: str

    def __lt__(self, other: "Row") -> bool:
        return (self.id, self.name) < (other.id, other.name)


class MockKeysetPaginator:
    """Pagination using keyset (seek) method instead of OFFSET."""…
14 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 Simulate Colocated Shard Joins in Python

Groups shards by their node and merges co-located shards into a single logical unit, checking capacity constraints.

sharding database distributed-systems
Python
import random
from collections import defaultdict


def simulate_colocated_shards_join(nodes: list[dict], shards: list[dict]) -> dict:
    """
    Simulates the join of co-located shards (on the same node) into a single
    logical shard. Returns the resulting node-to-shard mapping.

    Each node: {'id': str, 'capaci…
11 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
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 Shard Key Cardinality in Python

Generate mock data with configurable cardinality to evaluate shard key distribution and detect hotspots in database scaling design.

sharding cardinality database
Python
import random
import string

def calculate_cardinality(values):
    """Return the number of distinct values in the given list."""
    return len(set(values))

def generate_mock_data(num_records, cardinality):
    """Generate mock records for a shard key with given cardinality."""
    possible_keys = [f"key_{i:04d}" fo…
17 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.