Reference library

Python Code Samples

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

54 matches
Concurrency & performance medium

Build a Python Performance Profiler That Generates Readable Reports

Use cProfile and pstats to profile Python functions and print a sorted performance report showing the top time-consuming calls.

profiling cprofile pstats
Python
import cProfile
import pstats
import io
from pathlib import Path

def slow_function():
    total = 0
    for i in range(500_000):
        total += i ** 2
    return total

def fast_function():
    total = sum(i * i for i in range(500_000))
    return total

def profile_functions():
    profiler = cProfile.Profile()
  …
44 0 Open
Concurrency & performance medium

How to Profile CPU Hot Path in Python with cProfile and sort_stats cumtime

Profile a Python function's CPU usage by running cProfile, sorting stats by cumulative time, and printing a readable report to stdout.

cprofile profiling performance
Python
import cProfile
import pstats
import io


def slow_function():
    total = 0
    for i in range(100_000):
        total += i * i
    return total


def fast_function():
    return sum(i for i in range(100))


def main():
    slow_function()
    fast_function()


if __name__ == "__main__":
    profiler = cProfile.Profi…
12 0 Open
Concurrency & performance medium

How to Run Blocking Code in an Executor with asyncio in Python

This code runs blocking functions concurrently without stalling the event loop by offloading them to thread pool executors via asyncio.

asyncio executor concurrency
Python
import asyncio
import time


def blocking_task(name: str, duration: float) -> str:
    """Simulate a blocking operation."""
    time.sleep(duration)
    return f"Finished {name} after {duration}s"


async def main() -> None:
    loop = asyncio.get_running_loop()
    results = await asyncio.gather(
        loop.run_in_…
13 0 Open
Concurrency & performance medium

How to Use ProcessPoolExecutor for CPU Parallel Map in Python

Run a function over a sequence of inputs in parallel across multiple CPU cores with ProcessPoolExecutor.map.

concurrency processpoolexecutor parallelism
Python
from concurrent.futures import ProcessPoolExecutor
import math

def compute_square(num):
    return num * num

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

if __name__ == "__main__":
    numbers = rang…
11 0 Open
Concurrency & performance medium

How to Use multiprocessing Pool map and starmap in Python

Parallelize functions over iterables with Pool.map, and unpack multiple arguments via Pool.starmap.

multiprocessing parallelism pool
Python
from multiprocessing import Pool


def square(x):
    return x * x


def add_and_multiply(a, b, c):
    return (a + b) * c


if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]
    with Pool(processes=2) as pool:
        squares = pool.map(square, numbers)
        print(f"squares: {squares}")

        starmap_arg…
14 0 Open
Concurrency & performance medium

How to Use threading.RLock in Python

Demonstrates threading.RLock, a reentrant lock that allows the same thread to acquire it multiple times without deadlocking — essential for recursive functions sharing state across threads.

threading rlock concurrency
Python
import threading
import time

lock = threading.RLock()
shared_counter = 0

def recursive_increment(value, depth):
    global shared_counter
    with lock:
        shared_counter += 1
        print(f"Depth {depth}: counter = {shared_counter}")
        if depth > 1:
            recursive_increment(value, depth - 1)

def…
14 0 Open
Testing & modern typing medium

Characterization Test for Legacy Python Code

Capture the exact output of a legacy Python function for known inputs, creating a characterization test that documents current behavior before refactoring.

characterization-testing legacy-code testing
Python
def legacy_behavior(value):
    """Legacy function that returns a tuple with unconventional types."""
    if value == "special":
        return None, "legacy-special"
    elif value > 100:
        return value, "large"
    elif value > 0:
        return value * 2, "positive-doubled"
    elif value == 0:
     …
14 0 Open
Testing & modern typing medium

How to Benchmark Python Code with pytest-benchmark and mocks

Use pytest-benchmark to measure function performance while combining Mock and patch for controlled test scenarios.

pytest benchmark mock
Python
import time
from unittest.mock import Mock, patch

import pytest
from pytest_benchmark.fixture import BenchmarkFixture


def heavy_operation(data: list[int]) -> int:
    """Simulates a CPU-bound operation."""
    return sum(x * x for x in data)


def test_heavy_operation_benchmark(benchmark: BenchmarkFixture) -> None:…
14 0 Open
Testing & modern typing medium

How to Compare Execution Speed Between Python Functions

Measure and compare the average execution time of multiple Python functions using a reusable benchmark helper with time.perf_counter.

performance benchmarking time
Python
import time
import random

def method_a(values):
    """Sort using built-in sorted."""
    return sorted(values)

def method_b(values):
    """Sort using list's sort method."""
    values_copy = values[:]
    values_copy.sort()
    return values_copy

def method_c(values):
    """Sort manually using bubble sort (slow,…
38 0 Open
Testing & modern typing medium

How to Snapshot Test JSON with Mock in Python

Use pytest-snapshot to capture the exact output of a JSON-loading function, with and without mocking json.loads, so future changes are automatically detected.

pytest snapshot mock
Python
import json
from unittest.mock import Mock, patch
import pytest


def load_config(data):
    config = json.loads(data)
    return {"host": config["host"], "port": config["port"]}


def test_load_config_snapshot(snapshot):
    mock_data = json.dumps({"host": "localhost", "port": 8080, "extra": "ignored"})
    result = …
14 0 Open
System design patterns medium

How to Build a Pipe and Filter Text Processing Chain in Python

A functional pipe-and-filter chain that transforms text through uppercase, whitespace normalization, number removal, stopword filtering, and file export.

pipeline text-processing functional
Python
import re
import sys


def pipe_filter_chain(stream):
    def uppercase(text):
        return text.upper()

    def strip_whitespace(text):
        return " ".join(text.split())

    def remove_numbers(text):
        return re.sub(r"\d+", "", text)

    def remove_stopwords(text, stopwords={"the", "and", "of", "in"}):…
16 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}…
14 0 Open
Streaming & messaging medium

How to Mock Offset Commit Auto vs Manual in Python

Demonstrates a Kafka-style offset commit function with auto/manual modes and tests it using unittest.mock.patch.

unittest mocking kafka
Python
from unittest.mock import Mock, patch

def commit_offsets(topic_partition_offsets, auto_commit=False):
    """Manually commit offsets or simulate auto-commit."""
    if auto_commit:
        print(f"Auto-committing offsets: {topic_partition_offsets}")
        return {"status": "auto_committed"}
    
    print(f"Manuall…
15 0 Open
Caching & Redis medium

How to Cache Function Results in Redis with Python

A Python decorator that caches function results in Redis using TTL, with optional fakeredis for testing without a server.

redis caching decorator
Python
import redis
import json
import time
try:
    import fakeredis
except ImportError:
    fakeredis = None

from functools import wraps


def cache_redis(cache_key_prefix="cache", ttl=60):
    """Decorator to cache function results in Redis."""
    if fakeredis:
        r = fakeredis.FakeStrictRedis()
    else:
        r…
14 0 Open
Caching & Redis medium

How to Mock zlib Compression for Cache Values in Python

Compress cache values with zlib and mock the compress function in unit tests to simulate cache behavior.

zlib mock caching
Python
import zlib
from unittest.mock import patch

def compress_value(data: bytes) -> bytes:
    """Compress data using zlib and return the compressed bytes."""
    return zlib.compress(data)

def decompress_value(compressed: bytes) -> bytes:
    """Decompress zlib data and return the original bytes."""
    return zlib.deco…
11 0 Open
Caching & Redis medium

Implement a TTL cache with a mock clock in Python

This code creates a simple TTL cache that stores values with an expiration timestamp and allows injecting a mock time function to test expiry behavior deterministically.

cache ttl mocking
Python
import time
from functools import wraps

class TTLCache:
    def __init__(self, ttl_seconds):
        self.ttl = ttl_seconds
        self.cache = {}
        self._now = time.time

    def set_mock_time(self, mock_time_fn):
        """Inject a mock time function for testing TTL expiry."""
        self._now = mock_time_…
15 0 Open
Reliability & rate limiting medium

How to Implement Graceful Degradation with Feature Disabling in Python

A pattern that disables enhanced features and falls back to basic functionality when a dependency fails, with mock-based testing.

graceful-degradation feature-flags resilience
Python
import random
from unittest.mock import patch


class EnhancedFeature:
    """A feature that can gracefully degrade when a dependency is unavailable."""

    def __init__(self):
        self.feature_enabled = True

    def get_enhanced_data(self):
        """Simulate an enhanced feature that depends on external data."…
12 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:
…
15 0 Open
Reliability & rate limiting medium

Retry with Exponential Backoff and Jitter in Python

A decorator-style retry wrapper that retries a flaky function with exponential backoff plus random jitter, then raises after the last attempt fails.

retry backoff jitter
Python
import random
import time

def retry_with_backoff(func, max_retries=3, base_delay=0.5, max_jitter=0.1):
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries:
                raise
            delay = base_delay * (2 ** at…
14 0 Open
Microservices patterns medium

Distributed tracing with contextvars in Python

Propagate trace and span IDs across function calls using contextvars to mock distributed tracing in a single process.

tracing contextvars microservices
Python
import contextvars
import uuid
import time

_trace_context = contextvars.ContextVar("trace_context", default=None)


class TraceContext:
    def __init__(self, trace_id, parent_span_id):
        self.trace_id = trace_id
        self.parent_span_id = parent_span_id
        self.span_id = uuid.uuid4().hex[:16]
        s…
13 0 Open
Microservices patterns medium

Fallback cached response mock in Python

Wraps a mock function with a fallback to a real service and caches results to mask transient failures.

microservices caching fallback
Python
import time
from functools import wraps

class CachedMock:
    def __init__(self, cache_ttl=5):
        self.cache = {}
        self.cache_ttl = cache_ttl

    def get(self, key):
        cached = self.cache.get(key)
        if cached and time.time() - cached["timestamp"] < self.cache_ttl:
            return cached["v…
14 0 Open
Big data & Spark medium

How to Implement row_number Window Function in Python

This code implements a SQL-style ROW_NUMBER() window function in pure Python, partitioning rows by a set of columns and ranking them within each partition by an ordered set of columns.

window-functions data-processing row-number
Python
from collections import defaultdict
import itertools


def row_number(rows, partition_by, order_by):
    partitions = defaultdict(list)
    for index, row in enumerate(rows):
        key = tuple(row[col] for col in partition_by)
        partitions[key].append((index, row))

    result = []
    for key in partitions:
 …
17 0 Open
Big data & Spark medium

How to Mock a UDAF Aggregate Function in Python

This code provides a minimal mock of a User-Defined Aggregate Function (UDAF), simulating the initialize-update-merge-finalize lifecycle with a defaultdict counter.

udaf aggregate mock
Python
from collections import defaultdict

class MockUDAF:
    """A minimal mock of a User-Defined Aggregate Function.

    Simulates aggregate lifecycle: initialize, update per row,
    and finalize the result.
    """

    def __init__(self):
        self._buffer = defaultdict(int)

    def initialize(self):
        """Re…
13 0 Open
Big data & Spark medium

Lazy Evaluation Transform Lineage Mock in Python

Build a mock lineage tracker for data transforms using lazy evaluation and function wrappers in Python.

lazy-evaluation lineage decorator
Python
import functools


def lazy_transform(pipeline):
    """Build a mock lineage tracker using lazy evaluation."""
    lineage = []

    def wrap(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            lineage.append({"transform": func.__name__, "a…
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.