Reference library

Python Code Samples

Easy snippets you can copy, study, and run in the browser editor.

385 matches
Reliability & rate limiting easy

Rate Limiting with a Simple Python RateLimiter Class

A beginner-friendly Python rate limiter that tracks call timestamps and enforces a maximum number of calls within a rolling time window, with a helper to validate positive integers.

rate-limiting time api
Python
import time

class RateLimiter:
    def __init__(self, max_calls, period_seconds):
        self.max_calls = max_calls
        self.period_seconds = period_seconds
        self.calls = []

    def is_allowed(self):
        now = time.time()
        while self.calls and now - self.calls[0] >= self.period_seconds:
      …
14 0 Open
Observability & SRE easy

Check if a Timestamp Falls in a Daily Maintenance Window in Python

A small Python function that returns True when a datetime falls inside a daily maintenance window, and a demo printing yes/no for sample timestamps.

maintenance datetime scheduling
Python
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo


def in_maintenance_window(now: datetime, start_hour: int = 2, duration_hours: int = 4) -> bool:
    """Return True if 'now' falls inside the daily maintenance window."""
    day_start = now.replace(hour=start_hour, minute=0, second=0, microsecond…
19 0 Open
Observability & SRE easy

Generate Prometheus Text Exposition Format in Python

Mock a Prometheus metrics endpoint by formatting metrics into the text exposition format with HELP, TYPE, and sample lines.

prometheus metrics observability
Python
import time
from random import randint

# Mock a Prometheus metrics endpoint output
metrics = {
    "http_requests_total": {
        "help": "Total number of HTTP requests",
        "type": "counter",
        "samples": [
            {"labels": {"method": "get", "code": "200"}, "value": randint(1000, 9999)},
         …
17 0 Open
Observability & SRE easy

How to Mock an OTLP HTTP Endpoint in Python

This code implements a lightweight HTTP server that accepts OTLP/HTTP trace exports, stores spans by trace ID, and exposes them via a simple GET endpoint for debugging.

otlp http mock
Python
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from collections import defaultdict

class TraceHandler(BaseHTTPRequestHandler):
    traces = defaultdict(list)

    def do_POST(self):
        if self.path == "/v1/traces":
            length = int(self.headers.get("Content-Length", 0))
 …
14 0 Open
Observability & SRE easy

How to Redact Secrets from Log Messages in Python

Build a lightweight RedactingFormatter class that replaces sensitive tokens like passwords and API keys with [REDACTED] before log messages are printed.

redaction logging secrets
Python
class RedactingFormatter:
    def __init__(self, secrets):
        self.secrets = secrets

    def redact(self, message):
        for secret in self.secrets:
            message = message.replace(secret, "[REDACTED]")
        return message

    def format(self, record):
        message = record["message"]
        ret…
15 0 Open
Observability & SRE easy

How to Ship Logs to an Aggregator Endpoint in Python

Ship batched log entries to a mock HTTP aggregator endpoint with proper error handling and response status.

logging requests json
Python
import json
import requests
from datetime import datetime, timezone

LOG_ENTRIES = [
    {"timestamp": "2024-01-15T10:00:00Z", "level": "INFO", "message": "Server started"},
    {"timestamp": "2024-01-15T10:00:05Z", "level": "WARN", "message": "High memory usage"},
    {"timestamp": "2024-01-15T10:00:10Z", "level": "E…
14 0 Open
Observability & SRE easy

Mock Health Endpoint Liveness Check in Python

Simulate a liveness endpoint that reports service health with a configurable failure rate and uptime.

health check mock observability
Python
import time
import random


def liveness_check(service_name: str, failure_rate: float = 0.1) -> dict:
    """Mock health check that returns liveness status with a configurable failure rate."""
    healthy = random.random() > failure_rate
    response = {
        "service": service_name,
        "status": "alive" if he…
18 0 Open
Observability & SRE easy

Mocking a Metrics Gauge's set_value Method in Python

Demonstrates using unittest.mock.Mock with wraps to intercept a gauge's set_value call while verifying arguments and preserving real behavior.

unittest mocking metrics
Python
from unittest.mock import Mock

class MetricsGauge:
    def __init__(self, name):
        self.name = name
        self.value = 0.0

    def set_value(self, new_value):
        self.value = float(new_value)
        return self.value

# Usage demonstration with a mock
gauge = MetricsGauge("cpu_usage")
gauge_mock = Mock…
14 0 Open
Microservices patterns easy

BFF aggregation pattern: combine multiple service responses in Python

Mock three backend services and aggregate their responses into one unified payload — the BFF pattern every Python microservice gateway relies on.

bff aggregation microservices
Python
from dataclasses import dataclass
from typing import Any


@dataclass
class Service:
    name: str
    data: dict[str, Any]


def get_user_service() -> Service:
    return Service("user", {"id": 1, "name": "Alice"})


def get_orders_service() -> Service:
    return Service("orders", {"total": 299.99, "count": 2})


de…
17 0 Open
Microservices patterns easy

Cache-Aside Pattern in Python: Per-Service Mock

A Python mock of the cache-aside pattern for a single microservice—lazy-load from a database into an in-memory cache and invalidate on updates.

caching microservices cache-aside
Python
class ServiceCache:
    def __init__(self):
        self.database = {"user:1": "Alice", "user:2": "Bob", "user:3": "Charlie"}
        self.cache = {}

    def get_user(self, user_id):
        cache_key = f"user:{user_id}"
        if cache_key in self.cache:
            print(f"CACHE HIT: {cache_key}")
            retu…
16 0 Open
Microservices patterns easy

How to Build an In-Memory Service Registry Mock in Python

A simple in-memory ServiceRegistry class to register, retrieve, list, and unregister microservice endpoints or configs using a dict, with KeyError guards.

service-registry microservices in-memory
Python
class ServiceRegistry:
    def __init__(self):
        self._services = {}

    def register(self, name, service):
        self._services[name] = service

    def unregister(self, name):
        if name not in self._services:
            raise KeyError(f"Service '{name}' not found")
        del self._services[name]

 …
15 0 Open
Microservices patterns easy

How to Check an External Gateway vs Use an Internal Mock in Python

This code checks whether an external network gateway is reachable using ping, then falls back to a deterministic internal mock for testing environments.

network-check mock microservices
Python
import subprocess
import sys

def check_external_gateway():
    """True if we can reach an external network target."""
    try:
        subprocess.run(
            ["ping", "-c", "1", "-W", "2", "8.8.8.8"],
            capture_output=True,
            timeout=3,
            check=True,
        )
        return True
  …
16 0 Open
Microservices patterns easy

How to Mock a Schema Registry Avro Record in Python

Encode a Python dict into Avro binary using an inline schema, mimicking a schema registry record for tests or mocks.

avro schema-registry serialization
Python
import io
from avro.schema import parse
from avro.io import DatumWriter, BinaryEncoder

schema_json = """
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}
"""

schem…
16 0 Open
Microservices patterns easy

How to Mock an Ambassador Edge Proxy in Python

Build a lightweight mock Ambassador edge proxy with Python's http.server that responds to health and user endpoint requests for local development and testing.

ambassador mock http-server
Python
import http.server
import json
import urllib.parse
import threading

class AmbassadorProxyHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path == "/health":
            self.send_response(200)
            self.send_header("Content-T…
14 0 Open
Microservices patterns easy

How to Use the Adapter Pattern to Mock a Legacy System in Python

This code demonstrates the Adapter pattern, allowing a modern interface to interact with a legacy system by wrapping its outdated method.

adapter-pattern design-patterns legacy
Python
class LegacySystem:
    def legacy_method(self, data):
        return f"Legacy processed: {data}"

class ModernInterface:
    def process(self, data):
        raise NotImplementedError

class Adapter(ModernInterface):
    def __init__(self, legacy):
        self.legacy = legacy

    def process(self, data):
        re…
15 0 Open
Microservices patterns easy

How to mock an external service in Python with an anti-corruption facade

This code implements an anti-corruption facade that mocks an external API, allowing client code to interact with a simulated service while keeping the same interface.

microservices testing mocking
Python
class AntiCorruptionFacade:
    """Mocks a real API while keeping the same interface."""
    
    def __init__(self, data_store):
        self._data_store = data_store
        self._calls = []
    
    def get_user(self, user_id):
        self._calls.append(f"get_user({user_id})")
        return self._data_store.get(u…
12 0 Open
Microservices patterns easy

Mock a Sidecar Logger with Python Metrics

Simulate a sidecar logger that tracks request counts, error rates, and endpoint hits, producing a metrics snapshot.

microservices monitoring metrics
Python
import random
import time
from collections import defaultdict


class SidecarLogger:
    def __init__(self):
        self.metrics = defaultdict(int)
        self.total_requests = 0
        self.error_count = 0

    def log_request(self, endpoint, status_code):
        """Simulate logging a request and updating metrics…
17 0 Open
Big data & Spark easy

Compaction Small Files Mock in Python

Simulates a small-files compaction job by creating small mock files and merging them into a single output file using Python's standard library.

compaction file-io mock
Python
from pathlib import Path
import tempfile
import os


def create_small_files(directory: Path, file_count: int = 5, lines_per_file: int = 3):
    """Create several small mock files with sample content."""
    directory.mkdir(exist_ok=True)
    for i in range(file_count):
        file_path = directory / f"part-{i:04d}.tx…
21 0 Open
Big data & Spark easy

How to Explode an Array Column in Python

This code demonstrates a mock explode operation that converts an array column into multiple rows, similar to Spark's explode function.

explode arrays pyspark
Python
import json 

def explode_array_column(data, column):
    """Mock explode: split array column into multiple rows."""
    exploded = []
    for row in data:
        values = row.get(column, [])
        for value in values:
            new_row = dict(row)
            new_row[column] = value
            exploded.append(n…
15 0 Open
Big data & Spark easy

How to Implement MapReduce Word Count in Python Using a Dict

Simulate a MapReduce word count pipeline in Python with a mock dict, splitting text into words, shuffling, and reducing to frequency counts.

mapreduce word-count dictionary
Python
def map_reduce_word_count(text: str) -> dict:
    """Simulate a MapReduce pipeline to count word frequencies."""
    # MAP phase: split into words and emit (word, 1) pairs
    mapped = []
    for word in text.lower().split():
        # Clean word of punctuation
        clean_word = ''.join(char for char in word if cha…
16 0 Open
Big data & Spark easy

How to Implement collect_list in Python

Group rows by a key and collect all corresponding values into a list — a pure-Python mock of Spark's collect_list aggregation.

collect_list aggregation grouping
Python
from collections import defaultdict

def collect_list(rows, key_field, value_field):
    grouped = defaultdict(list)
    for row in rows:
        grouped[row[key_field]].append(row[value_field])
    return dict(grouped)

if __name__ == "__main__":
    data = [
        {"dept": "sales", "emp": "alice"},
        {"dept"…
18 0 Open
Big data & Spark easy

How to Truncate Lineage Back to a Checkpoint in Python

Walks a linked list of lineage nodes upward to find the nearest checkpoint and returns that node, truncating the lineage.

lineage checkpoint linked-list
Python
class LineageNode:
    def __init__(self, name, parent=None, checkpoint=None):
        self.name = name
        self.parent = parent
        self.checkpoint = checkpoint

    def truncate_at_checkpoint(self):
        """Truncate lineage back to the last checkpoint."""
        current = self
        while current.check…
18 0 Open
Big data & Spark easy

Hudi Upsert Mock Copy on Write in Python

Simulates Apache Hudi's Copy-on-Write upsert behavior by merging update records into a deep copy of base records, replacing matches or appending new ones.

hudi upsert copy-on-write
Python
import copy
from typing import Dict, List, Any

def upsert_copy_on_write(base_records: List[Dict[str, Any]], updates: List[Dict[str, Any]], key_field: str = "id") -> List[Dict[str, Any]]:
    """Simulate Hudi Copy-on-Write upsert: merge updates into a copy of base records."""
    result = copy.deepcopy(base_records)
 …
15 0 Open
Big data & Spark easy

Partition Data by Hash Key Mod N in Python

Returns a partition index for a string key by hashing it with MD5 and taking modulo N, then groups sample keys into partitions.

hashing partitioning hashlib
Python
import hashlib


def partition_key(key: str, num_partitions: int) -> int:
    """Return partition index for key using MD5 hash mod N."""
    digest = hashlib.md5(key.encode()).hexdigest()
    return int(digest, 16) % num_partitions


if __name__ == "__main__":
    keys = ["alice", "bob", "carol", "dave", "eve"]
    nu…
13 0 Open

Browse by section

Each section groups closely related Python snippets.

Guide: free Python code samples library

Copy-ready Python snippets for learners and developers

PythonSkillset code samples are short, focused examples organised by topic and difficulty. Every snippet is server-rendered HTML — readable by search engines and easy to copy. Open any sample, read the notes, copy the code, then press Try in editor to run it in the browser with Pyodide.

How to use this library

  1. Pick a topic section — strings, lists, files, functions, and more
  2. Open a sample, read How it works, and copy the code block
  3. Run it in the IDE, tweak values, then take a related quiz or tutorial lesson

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.