Reference library

Python Code Samples

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

299 matches
Reliability & rate limiting easy

Build a queue-based admission control system in Python

Implement a simple bounded-queue admission controller that accepts or rejects incoming requests based on current queue capacity.

admission-control queue rate-limiting
Python
from collections import deque
import time


class AdmissionControl:
    """Simple admission control using a bounded queue.

    Requests arrive at the queue; they are admitted in FIFO order.
    If the queue is full, the incoming request is rejected.
    """

    def __init__(self, capacity: int):
        self.capacit…
20 0 Open
Reliability & rate limiting medium

Circuit breaker failure threshold count in Python

Track consecutive or time-windowed failures with a deque to open a circuit breaker and auto-recover to half-open after a cooldown.

circuit-breaker resilience deque
Python
from collections import deque
from time import time, sleep


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_time: float = 10.0):
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.failures: deque[float] = deque()
        self.st…
17 0 Open
Reliability & rate limiting medium

How to Implement a Circuit Breaker in Python

A Python dataclass that provides circuit breaker logic with closed, open, and half-open states to fail fast on repeated errors.

circuit-breaker resilience fault-tolerance
Python
from dataclasses import dataclass
from datetime import datetime, timedelta
import time


@dataclass
class CircuitBreaker:
    failure_threshold: int = 3
    timeout_seconds: float = 5.0
    failures: int = 0
    state: str = "closed"
    last_failure: datetime = None

    def call(self, func):
        if self.state ==…
16 0 Open
Reliability & rate limiting medium

How to Mock a Circuit Breaker Reset Timeout in Python

This code implements a simple circuit breaker with a reset timeout test, simulating a flaky service to show half-open state transitions.

circuit-breaker reliability mock-testing
Python
import time
import random


class CircuitBreaker:
    def __init__(self, failure_threshold=3, reset_timeout=5):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED (nor…
16 0 Open
Reliability & rate limiting medium

How to Propagate Context Variables with asyncio in Python

Use Python's ContextVar with asyncio to carry deadline information across concurrent tasks and propagate context automatically.

contextvars asyncio concurrency
Python
import asyncio
from contextvars import ContextVar
from datetime import datetime

deadline = ContextVar("deadline", default=None)

async def worker(name):
    current = deadline.get()
    if current:
        print(f"{name} sees deadline: {current}")
    else:
        print(f"{name} sees no deadline")
    await asyncio.…
13 0 Open
Reliability & rate limiting easy

How to Retry on Specific Exception Tuples in Python

A decorator-based retry pattern that retries a function only when it raises exceptions specified in a tuple, with configurable retries and delay.

retry decorator exceptions
Python
import time
import random
from unittest.mock import patch


def retry_on_exceptions(retries=3, exceptions=(ValueError,), delay=0.1):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for attempt in range(retries):
                try:
                    return func(*args, **kwargs)
          …
16 0 Open
Reliability & rate limiting medium

Implement a Circuit Breaker Pattern in Python

This code implements a simple circuit breaker that opens after a threshold of consecutive failures, causing subsequent calls to fail fast without invoking the underlying function.

circuit-breaker reliability resilience
Python
class CircuitBreaker:
    def __init__(self, failure_threshold=3):
        self.failure_threshold = failure_threshold
        self.failure_count = 0
        self.open = False

    def call(self, func, *args, **kwargs):
        if self.open:
            raise RuntimeError("Circuit is open - failing fast")
        try:
…
16 0 Open
Reliability & rate limiting medium

Mock a Two-Phase Commit Coordinator in Python

Simulates a two-phase commit protocol where a coordinator asks participants to prepare, then commits or aborts based on unanimous readiness.

two-phase commit distributed systems transactions
Python
import random
import time
from typing import Dict, List


class TwoPhaseCommitCoordinator:
    def __init__(self, participants: List[str]):
        self.participants = participants
        self.participant_state: Dict[str, bool] = {}

    def prepare(self) -> bool:
        print("[Coordinator] Phase 1: Prepare")
     …
13 0 Open
Observability & SRE easy

How to Add Metadata Attributes to a Span in Python

Create a lightweight dataclass-based Span mock that stores key-value metadata attributes for tracing or event logging.

dataclasses observability tracing
Python
from dataclasses import dataclass, field
from typing import Dict, Any

@dataclass
class Span:
    name: str
    attributes: Dict[str, Any] = field(default_factory=dict)
    
    def set_attribute(self, key: str, value: Any) -> None:
        self.attributes[key] = value
    
    def get_attribute(self, key: str) -> Any…
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 Check Service Readiness Dependencies in Python

This code simulates a readiness check for external dependencies (database, cache, queue) with mock availability data and reports readiness status.

readiness dependencies health-check
Python
import sys
from datetime import datetime


def check_dependencies(config):
    results = []
    for dep, required in config.items():
        available = mock_availability(dep)
        status = "READY" if available >= required else "NOT READY"
        results.append((dep, available, required, status))
    return result…
12 0 Open
Observability & SRE medium

How to Check Uptime with a Synthetic HTTP Mock in Python

Run a mock HTTP server locally and probe it with urllib to measure synthetic uptime and response times, perfect for testing monitoring logic without external dependencies.

uptime http-server monitoring
Python
import http.server
import threading
import time
import urllib.request


class MockHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
   …
15 0 Open
Observability & SRE medium

How to Create a Mock OpenTelemetry Trace in Python

Create a mock OpenTelemetry trace in memory to test span creation, attributes, and parent-child relationships without exporting to a backend.

opentelemetry tracing testing
Python
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter


def create_mock_trace():
    tracer_provider = TracerProvider()
    span_exporter =…
14 0 Open
Observability & SRE medium

How to Create a TCP DNS Mock Server in Python

This code creates a mock TCP DNS server that listens on a specified port, accepts probe connections, and returns a fixed DNS response header to simulate a live DNS service for testing and observability.

socket dns tcp
Python
import socket
import threading


def handle_client(client_socket, address):
    print(f"[+] Connection from {address}")
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            print(f"[*] Received {len(data)} bytes (TCP DNS probe)")
          …
17 0 Open
Observability & SRE easy

How to Generate and Propagate W3C Trace Context Headers in Python

Generate and propagate W3C traceparent and tracestate headers for distributed tracing in Python, with mock service headers.

observability tracing w3c
Python
import uuid


def generate_w3c_traceparent(trace_id=None, parent_id=None, flags="01"):
    if trace_id is None:
        trace_id = uuid.uuid4().hex[:32]
    if parent_id is None:
        parent_id = uuid.uuid4().hex[:16]
    return f"00-{trace_id}-{parent_id}-{flags}"


def create_mock_headers(service_name, trace_id=N…
15 0 Open
Observability & SRE medium

How to Group Alerts by Time Window in Python

Group alert occurrences that fall within a sliding time window per alert key, reducing noise and summarizing bursts into single events.

alerts grouping monitoring
Python
from collections import defaultdict
from datetime import datetime, timedelta

def group_alerts(alerts, window_minutes=10):
    """Group alerts that occur within the same time window."""
    alerts_by_key = defaultdict(list)
    
    for alert in alerts:
        key = alert["key"]
        timestamp = alert["timestamp"]…
13 0 Open
Observability & SRE medium

How to Implement an Error Budget Policy in Python

This code implements a mock error budget policy that decides whether to freeze deployments based on a simulated error rate and monthly freeze limits.

error-budget sre deploy-freeze
Python
from datetime import datetime, timedelta

class ErrorBudgetPolicy:
    FREEZE_WINDOW_HOURS = 24
    MAX_FREEZES_PER_MONTH = 3

    def __init__(self, budget_percentage=99.9):
        self.budget_percentage = budget_percentage
        self.freeze_count = 0
        self.last_freeze_start = None
        self.freeze_enabl…
14 0 Open
Observability & SRE easy

How to Mock a Baggage Context (Key-Value Store) in Python

This code implements an in-memory key-value mock of a baggage context, letting you set, get, check, and delete keys for tracing-style metadata.

baggage tracing mock
Python
class BaggageContext:
    def __init__(self):
        self._store = {}

    def set(self, key, value):
        self._store[key] = value
        return value

    def get(self, key, default=None):
        return self._store.get(key, default)

    def has(self, key):
        return key in self._store

    def delete(sel…
16 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 Model Span Events in Python

Define a Span class with timestamped milestone events and a completion marker to track operation lifecycle.

observability dataclasses tracing
Python
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import List


class SpanStatus(Enum):
    STARTED = "started"
    COMPLETED = "completed"


@dataclass
class SpanEvent:
    name: str
    timestamp: float = field(default_factory=time.time)
    attributes: dict = field(default_facto…
15 0 Open
Observability & SRE easy

How to Route Alerts by Severity in Python

Map alert severity levels to routing targets and simulate dispatching alerts to on-call pages, email, Slack, or logs.

observability alerts routing
Python
def main():
    # Severity levels with corresponding alert routing targets
    routing_map = {
        "critical": "call_page",
        "high": "call_page",
        "medium": "email_team",
        "low": "slack_channel",
        "info": "log_only"
    }

    # Simulated alerts with severity
    alerts = [
        {"na…
13 0 Open
Observability & SRE easy

How to Simulate Trace Sampling Head in Python

Simulate head-based probabilistic trace sampling on mock trace data with a configurable sample rate and optional seed for reproducibility.

tracing sampling observability
Python
import random

def trace_sampling_head(mock_traces, sample_rate=0.5, seed=None):
    """Simulate probabilistic trace sampling (head-based) on mock data.
    
    Args:
        mock_traces: list of trace dictionaries with a unique 'trace_id'
        sample_rate: float 0.0-1.0, probability of keeping a trace
        see…
13 0 Open
Observability & SRE medium

Summary Quantile Mock Sketch in Python

Build a memory-efficient sketch that stores sorted bins of data points to answer approximate quantile queries like median without keeping all values in memory.

quantile sketch statistics
Python
import random
import statistics
from collections import Counter

class SummaryQuantileSketch:
    """
    A simple sketch that stores a fixed-size summary of data (min, max, deciles)
    using sorted bins, then answers approximate quantile queries.
    """
    def __init__(self, bins=10):
        self.bins = bins
    …
14 0 Open
Observability & SRE easy

Track Success Rates and Latency in Python: SRE Metrics Helper

A beginner-friendly Python class to record request outcomes and latencies, then report success rate, average latency, and p99.

sre metrics latency
Python
import random
import time
from collections import defaultdict


class MetricsTracker:
    """Simple helper to track success rates and latencies for SRE beginners."""

    def __init__(self):
        self.successes = 0
        self.failures = 0
        self.latencies = []

    def record(self, success, latency_ms):
   …
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.