Reference library

Python Code Samples

Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.

227 matches
Streaming & messaging medium

Mock Watermark Late Event Side Output in Python

Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.

watermark streaming side output
Python
from datetime import datetime, timedelta
from typing import List, Tuple


def watermark_mock(
    events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
    """Simulate watermarking: events arriving on time vs. late by ch…
12 0 Open
Streaming & messaging easy

Sliding Window Average with Deque in Python

Computes the running average of a sliding window over streaming numbers using a collections.deque for O(1) pop-left operations.

sliding-window deque streaming
Python
from collections import deque

class SlidingAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque()
        self.total = 0

    def add(self, value):
        self.window.append(value)
        self.total += value
        if len(self.window) > self.window_size:
…
14 0 Open
Caching & Redis medium

Cache Stampede Prevention with SingleFlight in Python

Implements a SingleFlight pattern in Python to deduplicate concurrent cache-miss computations and prevent cache stampede.

caching concurrency singleflight
Python
import threading
import time
from functools import wraps


class SingleFlight:
    def __init__(self):
        self._lock = threading.Lock()
        self._inflight = None

    def do(self, key, fn):
        with self._lock:
            if self._inflight is not None:
                return self._inflight[1]
           …
18 0 Open
Caching & Redis medium

How to Implement a Write-Through Cache in Python with a Mock Database

A thread-safe write-through cache that updates both cache and mock database atomically, computing values only after a successful write to the database.

caching write-through threading
Python
import threading
import time
import random


class WriteThroughCache:
    def __init__(self):
        self.cache = {}
        self.db = {}
        self.lock = threading.Lock()

    def write(self, key, value):
        with self.lock:
            # Simulate slow database write
            time.sleep(random.uniform(0.01…
13 0 Open
Caching & Redis easy

How to memoize a function in Python with lru_cache

Use functools.lru_cache to memoize a recursive Fibonacci function, caching results for a fixed number of calls to avoid repeated computation.

lru_cache memoization functools
Python
from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

if __name__ == "__main__":
    for i in range(10):
        print(f"fib({i}) = {fibonacci(i)}")
    print(f"Cache info: {fibonacci.cache_info()}")
14 0 Open
Caching & Redis medium

Implement a Multi-Level Cache with L1 Memory and L2 Redis in Python

This code implements a simple multi-level cache with an in-process L1 cache (via functools.lru_cache) and a mock Redis L2 cache with TTL, falling back to a slow computation on misses.

cache redis lru_cache
Python
import time
from functools import lru_cache


class MockRedis:
    def __init__(self):
        self.store = {}

    def get(self, key):
        return self.store.get(key, None)

    def set(self, key, value, ttl=5):
        self.store[key] = (value, time.time() + ttl)

    def get_ttl(self, key):
        value, expiry…
16 0 Open
Observability & SRE easy

Calculate Error Rate from Log Stream in Python

Parses a mock log stream to count errors and compute the error percentage using a rolling window of recent entries.

logging regex error-rate
Python
import re
from collections import deque

def error_rate_from_log_stream(message):
    log_pattern = r'^\[(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\] (ERROR|INFO|DEBUG): (.*)$'
    recent_entries = deque(maxlen=100)
    error_count = 0
    total_count = 0

    for line in message.strip().split('\n'):
        match = re.mat…
18 0 Open
Observability & SRE easy

Generate Mock CPU and Memory Metrics in Python

Build a mock_host_metrics() generator that outputs realistic CPU and memory usage percentages for monitoring demos and tests.

mock metrics monitoring
Python
import time
import random


def mock_host_metrics():
    """Generate mock CPU and memory metrics for a host."""
    cpu_percent = round(random.uniform(10.0, 95.0), 1)
    memory_percent = round(random.uniform(20.0, 90.0), 1)
    memory_used_mb = round(random.uniform(512, 8192), 1)

    return {
        "timestamp": in…
17 0 Open
Observability & SRE easy

Generate Synthetic CPU Utilization Metrics in Python

Creates realistic time-series CPU utilization samples with timestamps, noise, and output as structured JSON for observability demos and testing.

observability metrics time-series
Python
from datetime import datetime, timedelta
import random
import json


def generate_metric_samples(base_value, noise, count=60, interval_minutes=1):
    """Generate realistic CPU utilization samples for a given time window."""
    timestamps = []
    values = []

    now = datetime.utcnow()
    start_time = now - timede…
15 0 Open
Observability & SRE easy

Generate Synthetic SRE Metrics and Calculate Availability in Python

Create realistic service metrics with random latency, error rate, and request counts, then compute availability and summarize the stream for SLO checks.

sre synthetic-data metrics
Python
from datetime import datetime, timedelta
import random

def generate_service_metrics(service_name: str, minutes: int = 30) -> list[dict]:
    """Generate synthetic SRE metrics for a service across recent minutes."""
    metrics = []
    now = datetime.now()
    
    for i in range(minutes):
        timestamp = now - t…
15 0 Open
Observability & SRE medium

How to Build a Python Latency Histogram with Mock Buckets

This code implements a mock latency histogram that records request durations into configurable buckets and outputs counts, total, and average latency.

histogram latency metrics
Python
import time
import random
from collections import Counter


class LatencyHistogram:
    def __init__(self, buckets):
        self.buckets = sorted(buckets)
        self.counts = Counter()
        self.total = 0
        self.sum_latency = 0

    def record(self, latency_ms):
        for i, boundary in enumerate(self.bu…
14 0 Open
Observability & SRE easy

How to Calculate Apdex Score from Latency Data in Python

Generate simulated latency samples and compute the Apdex score to gauge user satisfaction with an application's performance.

apdex latency observability
Python
import random
import statistics

def generate_latencies(count=100, base=100, stddev=30):
    return [max(0, random.gauss(base, stddev)) for _ in range(count)]

def apdex(latencies, threshold=200):
    satisfied = sum(1 for lat in latencies if lat < threshold)
    tolerating = sum(1 for lat in latencies if lat >= thres…
16 0 Open
Observability & SRE easy

How to Calculate Percentile Latency in Python

Generate mock latency samples with occasional spikes and compute 50th, 90th, 95th, and 99th percentile values in milliseconds.

percentile latency slo
Python
import random
import statistics

def generate_latency_samples(n=1000):
    """Generate realistic mock latency data (ms) with occasional spikes."""
    samples = []
    for _ in range(n):
        # Normal case: ~50ms with jitter
        base = random.gauss(50, 5)
        # 2% spike chance: slow downstream or GC pause
 …
15 0 Open
Observability & SRE easy

How to Calculate SLO Error Budget in Python

Simulate an SLO error budget by computing allowed downtime from a target availability percentage and mocking monthly incidents.

slo error-budget monitoring
Python
```python
import random


def calculate_error_budget(total_seconds: int, target_availability: float) -> float:
    return (1.0 - target_availability) * total_seconds


def simulate_monthly_availability(seconds_in_month: int, budget_seconds: float) -> float:
    # Mock: randomly consume a fraction of the error budget i…
16 0 Open
Observability & SRE easy

How to Compute SRE Metrics Like Error Rate and Availability in Python

Tracks log events in a sliding time window and calculates error rate per second and availability percentage using an easy-to-follow class.

observability sre metrics
Python
from collections import deque
from datetime import datetime, timedelta
from typing import Dict, Deque


class LogMetrics:
    """Simple observability helper to track log events and calculate SRE metrics."""

    def __init__(self, window_seconds: int = 60):
        self.window_seconds = window_seconds
        self.eve…
15 0 Open
Observability & SRE easy

How to Do Structured JSON Logging in Python

Create a custom logging formatter that outputs each log entry as a single JSON line with timestamp, level, logger name, and message.

logging json observability
Python
import json
import logging
from datetime import datetime


class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.ge…
16 0 Open
Observability & SRE easy

How to Process System Metrics (RSS, CPU) in Python

Simulate and aggregate RSS and CPU system metrics to compute averages and maximums for monitoring dashboards.

metrics rss cpu
Python
import random
import time
from collections import namedtuple

Metric = namedtuple("Metric", ["name", "value", "unit"])


def generate_metrics(num_metrics: int = 5) -> list:
    """Simulate a batch of system metrics."""
    metrics = []
    for i in range(num_metrics):
        rss = random.randint(50, 500)  # MB
      …
13 0 Open
Observability & SRE medium

How to Track Cache Hit Ratio in Python

Simulate an LRU cache with hit/miss tracking and compute a real-time hit ratio from random access patterns.

cache lru hit-ratio
Python
import random
import time
from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity
        self.hits = 0
        self.misses = 0

    def get(self, key):
        if key in self.cache:
            self.hits += 1
     …
14 0 Open
Observability & SRE easy

How to Use Log Levels DEBUG INFO WARNING ERROR in Python

Demonstrates Python's logging levels (DEBUG, INFO, WARNING, ERROR) with basicConfig and a logger, showing how severity filtering controls output.

logging log-levels observability
Python
import logging

# Configure a mock logger to demonstrate log levels
logging.basicConfig(level=logging.DEBUG, format="%(levelname)s: %(message)s")
logger = logging.getLogger("mock_logger")

# Simulate events at each severity level
logger.debug("Detailed diagnostic info")
logger.info("General system operation")
logger.w…
14 0 Open
Observability & SRE easy

How to mock SLI availability success ratio in Python

Simulate request outcomes with deterministic randomness and compute the SLI availability success ratio to check if a target is met.

sli availability monitoring
Python
import random
from collections import defaultdict

def mock_availability(num_requests=1000, target_ratio=0.995):
    """
    Simulate request outcomes and compute the SLI availability success ratio.
    
    Args:
        num_requests: Total number of requests to simulate
        target_ratio: Target availability rati…
14 0 Open
Microservices patterns easy

How to Build a Microservice Helper in Python

A beginner-friendly Python helper that validates input, normalizes service responses, and simulates user management—showing clean patterns for microservice development.

microservices validation oop
Python
import json
from typing import Any, Dict, List


class DataValidator:
    """Simple validator for common data patterns."""

    @staticmethod
    def is_valid_email(value: str) -> bool:
        """Check if value looks like an email."""
        return "@" in value and "." in value.split("@")[-1]

    @staticmethod
    …
13 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…
18 0 Open
Big data & Spark easy

How to Broadcast a Small Lookup Table in Python

Simulates broadcasting a small lookup table by iterating key-value pairs and emitting packed rows to subscribers with deterministic output.

broadcast lookup-table dictionary
Python
import random

# Generate a deterministic mock broadcast of a small lookup table
# with 5 keys and random integer values (seeded for reproducibility)

data = {
    "sensor_a": 22,
    "sensor_b": 87,
    "sensor_c": 43,
    "sensor_d": 65,
    "sensor_e": 31,
}

# Simulate a broadcast to subscribers by iterating and p…
16 0 Open
Big data & Spark medium

How to Build a DAG Execution Stage Calculator in Python

Computes the execution stages of a directed acyclic graph (DAG) by grouping nodes that become ready simultaneously using topological sorting with Kahn's algorithm.

dag topological-sort kahn-algorithm
Python
from collections import defaultdict, deque


def get_stages(edges):
    """Return list of stages, where each stage is a list of nodes
    that become ready at the same time in a DAG."""
    graph = defaultdict(list)
    in_degree = defaultdict(int)
    nodes = set()

    for src, dst in edges:
        graph[src].appen…
16 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.