Database scaling & optimization
Indexing, connection pooling, read replicas, query tuning, and throughput-aware SQL.
How to Replicate Data Across All Shards in Python
Mocks a global table that replicates a key-value pair to every shard, ensuring reads return the same value from any shard.
from dataclasses import dataclass
from typing import Dict, List
@dataclass
class Shard:
id: str
data: Dict[str, int]
class GlobalTable:
def __init__(self, shards: List[Shard]):
self._shards = {s.id: s for s in shards}
def set_value(self, key: str, value: int) -> None:
"""Replicate …
How to Shard Data by User ID Hash in Python
Deterministically map user IDs to shard indexes using an MD5 hash modulo the shard count in Python.
import hashlib
def shard_id(user_id: str, num_shards: int = 4) -> int:
"""Deterministically map a user_id to a shard index using MD5."""
digest = hashlib.md5(user_id.encode("utf-8")).hexdigest()
return int(digest[:8], 16) % num_shards
if __name__ == "__main__":
user_ids = ["alice", "bob", "carol", "d…
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.
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…
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.
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…
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.
```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 = …
How to Speed Up Column Lookups with DataFrame Index in Python
Use pandas set_index to make repeated column value lookups O(1)-style fast instead of scanning the whole DataFrame each time.
import pandas as pd
# Mock dataset with duplicate customer IDs
data = {"customer_id": [101, 102, 103, 101, 104, 102],
"order_amount": [250.0, 85.5, 300.0, 175.25, 420.0, 95.75]}
df = pd.DataFrame(data)
df = df.set_index("customer_id")
# Simulated lookup request
search_id = 102
# Fast index-based lookup (no…
How to Validate Data Before Scaling in Python
A reusable Python helper that validates required fields and constraint checks on data rows before entering a database pipeline, improving data quality and throughput.
def validate_data(data, required_fields, constraints=None):
"""
Basic validation helper demonstrating data-quality workflows
before scaling (catches bad rows early, improves throughput).
"""
constraints = constraints or {}
errors = []
for field in required_fields:
if field not in d…
How to enforce a unique index constraint in Python
Mock a database unique index in Python that rejects duplicate rows based on one or more columns.
class MockIndex:
def __init__(self, columns):
self.columns = columns
self._values = set()
def insert(self, row):
key = tuple(row[col] for col in self.columns)
if key in self._values:
raise ValueError(f"Duplicate key {key} for columns {self.columns}")
self._v…
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.
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 …
How to mock directory-based sharding in Python
Simulates distributing files into logical shards using a deterministic hash of each filename, mocking how a database might shard rows across nodes.
import os
import hashlib
from collections import defaultdict
from pathlib import Path
def get_shard_for_key(key: str, num_shards: int) -> int:
"""Return a deterministic shard index (0..num_shards-1) for a key."""
digest = hashlib.md5(key.encode('utf-8')).hexdigest()
return int(digest, 16) % num_shards
…
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.
"""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…
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.