Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
How to Simulate Timeout with Custom TimeoutError in Python
Run a function in a daemon thread and raise a custom TimeoutError if it exceeds a specified time limit.
import time
from typing import Callable, TypeVar
T = TypeVar("T")
class TimeoutError(Exception):
"""Raised when an operation exceeds its time limit."""
def __init__(self, message: str = "Operation timed out"):
self.message = message
super().__init__(self.message)
def run_with_timeout(func…Graceful Shutdown Executor Context Manager in Python
A context manager that starts a background thread and ensures it stops gracefully on exit, handling timeouts and exceptions.
import signal
import threading
import time
from contextlib import contextmanager
@contextmanager
def graceful_shutdown_executor(timeout=5.0):
"""Context manager that runs a task and gracefully stops it on timeout or exception."""
stop_event = threading.Event()
def task():
print("Task started")
…How to Demonstrate the GIL with Python Threads vs Processes
Measure and compare wall-clock time for CPU-bound work using Python threads (limited by the GIL) versus multiprocessing (which bypasses the GIL).
import threading
import multiprocessing
import time
import os
def cpu_heavy(n):
return sum(i * i for i in range(n))
def run_threads(n):
threads = [threading.Thread(target=cpu_heavy, args=(n,)) for _ in range(2)]
start = time.perf_counter()
for t in threads:
t.start()
for t in threads:
…How to Pause and Resume Threads with threading.Event in Python
Use threading.Event to pause and resume worker threads in Python, controlling execution flow with set and clear methods.
import threading
import time
workers = []
def worker(name, event):
for i in range(10):
event.wait()
print(f"{name} step {i}")
time.sleep(0.1)
def pause_worker(name):
global pause_event
for w in workers:
if w.name == name:
pause_event.clear()
print(…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.
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_…How to Speed Up Data Filtering with Python ThreadPoolExecutor
This code compares sequential filtering of even numbers with a threaded version using ThreadPoolExecutor, showing a measurable speedup for I/O-bound work.
import time
from concurrent.futures import ThreadPoolExecutor
import random
def is_even(number):
time.sleep(0.001) # simulate work
return number % 2 == 0
def filter_even_sequential(numbers):
return [n for n in numbers if is_even(n)]
def filter_even_threaded(numbers):
with ThreadPoolExecutor(max_…How to Use ThreadPoolExecutor for Concurrent Tasks in Python
Compare sequential execution with ThreadPoolExecutor for I/O-bound tasks, measuring speedup and timing with perf_counter.
import time
import threading
from concurrent.futures import ThreadPoolExecutor
def fetch_data(index):
"""Simulate a synchronous data fetch."""
time.sleep(0.1)
return f"data-{index}"
def run_sequential(total=10):
"""Run tasks one after another."""
start = time.perf_counter()
results = [fetch…How to Use a Bounded Buffer with threading.Condition in Python
Implement a thread-safe bounded buffer using threading.Condition and show a producer–consumer example with exact output.
import threading
import time
import random
class BoundedBuffer:
def __init__(self, capacity):
self.capacity = capacity
self.buffer = []
self.condition = threading.Condition()
def put(self, item):
with self.condition:
while len(self.buffer) >= self.capacity:
…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.
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…How to Use threading.local for Per-Thread Data in Python
Use threading.local to keep thread-specific data — each thread gets its own copy of the attribute, so values don't leak between threads.
import threading
import time
local_storage = threading.local()
def worker(name):
local_storage.name = name
time.sleep(0.1)
print(f"Thread {threading.current_thread().name}: {local_storage.name}")
if __name__ == "__main__":
threads = []
for i in range(3):
t = threading.Thread(target=worke…Thread Pool Map for IO Bound Tasks in Python
Run IO-bound mock tasks concurrently with ThreadPoolExecutor.map and measure total elapsed time in Python.
import concurrent.futures
import time
from pathlib import Path
def mock_io_task(filename):
"""Simulate an IO-bound task by creating a small file and measuring its latency."""
path = Path(filename)
path.write_text("data")
time.sleep(0.1) # Simulate slow disk/network
return f"{filename} written in …How to Limit Concurrent Requests with a Semaphore in Python
Use threading.Semaphore with a ThreadPoolExecutor to cap how many worker threads run simultaneously, preventing resource overload.
import threading
import time
from concurrent.futures import ThreadPoolExecutor
def worker(name, semaphore, results):
with semaphore:
results.append(f"start {name}")
time.sleep(0.5) # simulate async work
results.append(f"done {name}")
def main():
sem = threading.Semaphore(2) # max 2 …How to Mock a 202 Accepted Long-Running Operation in Python
Build a mock HTTP server that returns a 202 Accepted response immediately and simulates a long-running operation in the background with threading.
import time
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
class MockHandler(BaseHTTPRequestHandler):
def do_POST(self):
if self.path == "/long-running":
self.send_response(202)
self.send_header("Content-Type", "application/json")
self.end_h…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.
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)
…Simulate RabbitMQ QoS Prefetch Count in Python
Mocks RabbitMQ QoS prefetch semantics using threading and a queue to cap concurrent unacked message processing per worker.
import threading
import time
import queue
class RabbitMQMock:
def __init__(self, prefetch_count=1):
self.prefetch_count = prefetch_count
self.channel_queue = queue.Queue()
self.currently_processing = 0
self.lock = threading.Lock()
def start_consuming(self, messages, worker_co…Cache Stampede Prevention with SingleFlight in Python
Implements a SingleFlight pattern in Python to deduplicate concurrent cache-miss computations and prevent cache stampede.
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]
…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.
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…At Least Once with Idempotent Consumer in Python
Implements a thread-safe idempotent consumer that processes each unique message exactly once, even when a producer sends duplicates under an at-least-once delivery model.
import threading
import time
import uuid
from collections import Counter
class IdempotentConsumer:
def __init__(self):
self.processed = set()
self._lock = threading.Lock()
def consume(self, message_id, payload):
with self._lock:
if message_id in self.processed:
…How to Implement Hedged Requests in Python
This code demonstrates a hedged request pattern using threading, which sends duplicate calls and returns the first result that arrives within a timeout.
import time
from unittest.mock import Mock
def hedged_request(call, timeout=0.05):
"""Execute two duplicate calls, return first result within timeout."""
result_container = {}
def run_and_store():
result_container['result'] = call()
result_container['done'] = True
# Simulate slow cal…How to Implement a Bulkhead Pattern with Threading in Python
Implement a bulkhead pattern in Python that isolates concurrent tasks with a bounded semaphore, limiting active workers to prevent resource exhaustion.
import threading
import time
import random
class Bulkhead:
def __init__(self, workers: int):
self._semaphore = threading.BoundedSemaphore(workers)
self._lock = threading.Lock()
self._active = 0
def run(self, task):
with self._semaphore:
with self._lock:
…How to implement a rate-limited shared counter in Python
Implements a thread-safe global counter that allows a maximum number of increments per second using a lock and time-based refill.
import threading
import time
import random
counter = 0
lock = threading.Lock()
MAX_CALLS_PER_SECOND = 3
last_refill = time.time()
def rate_limited_increment():
global counter, last_refill
with lock:
now = time.time()
if now - last_refill >= 1.0:
last_refill = now
count…Mock Distributed Rate Limiter with Dict in Python
Simulates a distributed token-bucket rate limiter with a thread-safe dict, useful for testing before moving to Redis.
import time
import threading
from collections import defaultdict
class DistributedRateLimiter:
"""
A mock distributed rate limiter using a dict with thread-safe access.
Implements a token bucket algorithm per user.
"""
def __init__(self, rate_per_second=5, burst_capacity=10):
self.rate_p…Token bucket rate limiter in Python (in-memory)
Implement a thread-safe in-memory token bucket rate limiter that throttles requests based on a steady refill rate.
import time
import threading
class TokenBucket:
def __init__(self, capacity, refill_rate, refill_interval=1.0):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate
self.refill_interval = refill_interval
self.last_refill = time.monotonic()
…How to Mock Mutual Exclusion for A/B Experiment Groups in Python
Simulate mutual exclusion for experiment groups using a thread-safe lock, ensuring only one member updates the shared counter at a time.
import threading
import time
import random
class CountingGate:
"""A mock mutual exclusion gate using a lock."""
def __init__(self):
self.counter = 0
self.lock = threading.Lock()
def enter(self, group_id, member_id):
with self.lock:
current = self.counter
t…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
- Pick a topic section — strings, lists, files, functions, and more
- Open a sample, read How it works, and copy the code block
- 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.