Reference library

Python Code Samples

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

506 matches
System design patterns medium

How to Migrate a Legacy Facade with the Strangler Fig Pattern in Python

Use a facade to wrap a legacy API and incrementally migrate callers to a modern interface, following the strangler fig pattern.

facade legacy migration
Python
class LegacyAPI:
    """Simulates the legacy system's raw interface."""
    def get_user(self, user_id):
        return {"id": user_id, "name": "Alice", "legacy": True}


class UserService:
    """Facade that wraps the legacy system with a modern interface."""
    def __init__(self, legacy_api=None):
        self.lega…
14 0 Open
System design patterns easy

How to Mock a Metrics Decorator in Python with unittest.mock

This code demonstrates a timing decorator that wraps a function to measure execution time and prints the duration, with a unit test using unittest.mock to patch the print function and assert it was called.

decorators unittest.mock metrics
Python
import time
from functools import wraps
from unittest.mock import patch

def add_metrics(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.6f}s…
18 0 Open
System design patterns easy

How to mock the domain center in an onion architecture in Python

Define a repository interface and an in-memory mock to test domain services without touching infrastructure.

onion-architecture repository-pattern dependency-injection
Python
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, List, Optional


@dataclass
class Order:
    id: int
    customer: str
    items: List[str]
    total: float


class OrderRepository(ABC):
    @abstractmethod
    def find_by_id(self, order_id: int) -> Optional[Order]:
     …
13 0 Open
API design & gRPC medium

Build a Bulk Array POST Mock Server in Python

Creates an HTTP mock server that accepts POST requests with a JSON array and returns incremental IDs for each item.

http-server mock-api rest
Python
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse

class MockHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        if urlparse(self.path).path != "/bulk":
            self.send_response(404)
            self.end_headers()
            return

        cont…
20 0 Open
API design & gRPC medium

How to Build a Mock REST GET Endpoint Handler in Python

Create a lightweight mock REST GET server in Python using the standard library, with a dict-based route registry that maps paths to handler functions and returns JSON responses with proper HTTP status codes.

mock-server rest-api http
Python
from http.server import BaseHTTPRequestHandler, HTTPServer
import json

# Mock API handler registry
def handle_users():
    return {"status": "ok", "data": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}

def handle_products():
    return {"status": "ok", "data": [{"id": 101, "name": "Laptop", "price": 999.99}…
15 0 Open
API design & gRPC easy

How to Parse gRPC Request Data in Python

Build a beginner-friendly gRPC service handler that parses incoming protobuf messages into Python dictionaries and starts a simple gRPC server.

grpc protobuf api
Python
from google.protobuf import json_format
import grpc
from concurrent import futures
import time


class DataParsingService:
    def parse(self, request):
        return {
            "received_json": json_format.MessageToJson(request),
            "parsed_fields": {
                "name": request.name,
               …
17 0 Open
API design & gRPC easy

How to Poll an Operation Status Endpoint in Python

Mock a polling endpoint in Python that simulates checking an async operation's status until it completes or times out.

polling api async
Python
import time
import random


def poll_status(url: str, timeout: float = 5.0) -> dict:
    """Mock a polling endpoint that eventually returns a completed status."""
    start = time.time()
    while time.time() - start < timeout:
        # Simulate delayed response
        time.sleep(0.2)
        # 80% chance to report …
14 0 Open
API design & gRPC easy

How to Prefix Python API URIs with a Version Slug

Build a versioned API endpoint by optionally adding a version prefix like v1 to the URL path using the stdlib urllib module.

api url urllib
Python
from urllib.parse import urlparse

BASE_URL = "https://api.example.com"

def build_uri(resource, version="v1"):
    """Mock a versioned API URI with an optional v1 prefix."""
    parsed = urlparse(BASE_URL)
    prefix = f"/{version}" if version else ""
    return f"{parsed.scheme}://{parsed.netloc}{prefix}/{resource.l…
16 0 Open
API design & gRPC easy

How to mock a REST POST endpoint in Python

Create a simple mock REST server that responds to POST requests with a 201 status and a JSON body.

http mock api
Python
import json
from http.server import BaseHTTPRequestHandler, HTTPServer


class MockHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length) if content_length else b"{}"
        try:
            data = json…
13 0 Open
Streaming & messaging medium

How to Aggregate Periodic Snapshot Data in Python

Generates mock snapshot data and groups values into periods to compute average aggregates with Python's standard library.

aggregation snapshots streaming
Python
import random
from collections import defaultdict

def snapshot_aggregate(n=10, period=3):
    data = defaultdict(list)
    for i in range(n):
        key = f"item_{i % period}"
        data[key].append(random.randint(1, 100))
    return dict(data)

def aggregate_periodic(snapshots, period=3):
    result = {}
    for …
15 0 Open
Streaming & messaging medium

How to Implement Backpressure Pause Producer with a Bounded Queue in Python

Places a Producer thread that sends items into a bounded queue with backpressure: on Full, it pauses to let the consumer catch up.

queues backpressure threading
Python
import threading
import time
import queue
import random


class Producer:
    def __init__(self, q):
        self.q = q
        self.running = True

    def produce(self):
        while self.running:
            item = random.randint(1, 100)
            try:
                self.q.put(item, timeout=0.5)
              …
15 0 Open
Streaming & messaging easy

How to Partition and Order Kafka-Style Messages by Key in Python

Group messages with the same key into ordered buckets using hashing and a defaultdict, mimicking Kafka partition ordering.

streaming partitioning kafka-pattern
Python
from dataclasses import dataclass
from collections import defaultdict

@dataclass
class Message:
    key: str
    content: str

def partition_and_order(messages, num_partitions=3):
    partitions = defaultdict(list)
    for msg in messages:
        partition_id = hash(msg.key) % num_partitions
        partitions[parti…
15 0 Open
Streaming & messaging easy

How to Simulate a Micro-Batch Interval Trigger in Python

A dataclass-based mock that emits batch numbers at fixed intervals, mimicking a micro-batch streaming scheduler for testing and development.

streaming mock dataclass
Python
import time
from dataclasses import dataclass, field
from typing import List, Callable


@dataclass
class MicroBatchTriggerMock:
    batch_interval_seconds: float = 0.5
    max_batches: int = 5
    _batches_emitted: int = 0
    _next_emit_time: float = field(init=False, default=0)

    def start(self, on_batch: Callab…
17 0 Open
Streaming & messaging medium

How to Track Session Windows with Gap Timeout in Python

A Python class that groups events into sessions, closing a session when the gap between events exceeds a timeout threshold.

session-window streaming timeout
Python
import time

class SessionWindow:
    """Track sessions with a gap timeout (mock)."""
    
    def __init__(self, timeout_seconds=5):
        self.timeout = timeout_seconds
        self.session_start = None
        self.last_event_time = None
        self.event_count = 0
        self.events = []
    
    def add_event…
17 0 Open
Streaming & messaging easy

How to Wrap Message Attributes in a CloudEvent with Python

Create a minimal CloudEvent dataclass that wraps arbitrary message attributes into a JSON envelope, matching CloudEvents 1.0 spec.

cloudevents messaging dataclasses
Python
import json
from dataclasses import dataclass, field, asdict
from typing import Any, Dict
from datetime import datetime, timezone


@dataclass
class CloudEvent:
    message_attributes: Dict[str, Any] = field(default_factory=dict)

    def wrap(self, event_id: str, source: str, event_type: str, data: Any):
        self…
14 0 Open
Streaming & messaging medium

How to mock a CQRS projector read model update in Python

Build a CQRS projector class that maintains denormalized read models by applying domain events in a mock order-processing service.

cqrs projector read-model
Python
from dataclasses import dataclass, field
from typing import Dict, List, Optional


@dataclass
class OrderReadModel:
    order_id: str
    customer_name: str
    total: float
    status: str = "pending"
    items: List[Dict] = field(default_factory=list)

    def apply_event(self, event_type: str, payload: Dict) -> Non…
14 0 Open
Streaming & messaging medium

Mock Kafka Consumer Group Partition Assignment in Python

Simulates a Kafka consumer group's round-robin partition assignment with a Python class and prints assignments per consumer.

kafka consumer-group partition-assignment
Python
from collections import defaultdict


class ConsumerGroupAssignment:
    def __init__(self, group_name, topics_partitions):
        self.group_name = group_name
        self.consumers = {}
        self.assignments = defaultdict(set)
        topics_partitions = sorted(
            [(topic, partition) for topic, partiti…
15 0 Open
Streaming & messaging hard

Mock Protobuf Binary Encoding in Python

Demonstrates a minimal protobuf-like binary encoding and decoding of an event dataclass using varints and length-delimited fields in pure Python.

protobuf binary-encoding varint
Python
import struct
from dataclasses import dataclass


@dataclass
class Event:
    id: int
    user_id: int
    action: str

    def encode(self) -> bytes:
        # Mock protobuf-like binary encoding using varint and length-delimited fields
        buf = bytearray()
        # field 1: varint id (tag = (1 << 3) | 0 = 8)
  …
13 0 Open
Streaming & messaging easy

Using the retained message flag in MQTT with Python

This script subscribes to an MQTT topic and prints the retained flag for each received message, demonstrating how to distinguish retained messages from normal ones.

mqtt paho-mqtt iot
Python
import paho.mqtt.client as mqtt

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    # Subscribe to a topic and check retained flag
    client.subscribe("test/retained")
    print("Subscribed to test/retained")

def on_message(client, userdata, msg):
    # msg.retain is the M…
15 0 Open
Reliability & rate limiting medium

How to Implement an Adaptive Rate Limiter in Python

Build an adaptive rate limiter that adjusts request intervals dynamically based on recent error rates, slowing down when failures spike.

rate-limiting backoff adaptive
Python
import time
import random

class AdaptiveRateLimiter:
    """Simple adaptive rate limiter that reduces requests when error rate is high."""
    
    def __init__(self, min_interval=0.1, max_interval=2.0, error_threshold=0.3):
        self.min_interval = min_interval
        self.max_interval = max_interval
        sel…
13 0 Open
Reliability & rate limiting easy

How to Mock a Try Confirm Cancel Pattern in Python

Define a simple class with confirm and cancel methods, execute a try confirm with error handling, and print the final state.

try-except mock class
Python
class TCC:
    def __init__(self):
        self.confirmed = False
        self.cancelled = False

    def confirm(self):
        self.confirmed = True
        return "confirmed"

    def cancel(self):
        self.cancelled = True
        return "cancelled"

    def try_confirm(self):
        try:
            result =…
15 0 Open
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 medium

Adding a Correlation ID to Log Context in Python

Injects a correlation ID into the logging context using a context manager and a custom log record factory so every log line includes the ID.

logging correlation-id context-manager
Python
import logging
import uuid
from contextlib import contextmanager

logging.basicConfig(level=logging.INFO, format='%(levelname)s | %(correlation_id)s | %(message)s')


@contextmanager
def correlation_id_context(correlation_id):
    """Temporarily inject a correlation_id into the logging context."""
    extra = {'correl…
17 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

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.