Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
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…Limit Concurrency with asyncio.Semaphore in Python
Use asyncio.Semaphore to cap how many async tasks run at once, throttling a batch of coroutines to a set concurrency limit.
import asyncio
import random
async def fetch_data(i: int, semaphore: asyncio.Semaphore) -> str:
async with semaphore:
print(f"Task {i} starts")
await asyncio.sleep(random.uniform(0.1, 0.5))
print(f"Task {i} finishes")
return f"Result {i}"
async def main() -> None:
semaphore …Mocking Trio's open_nursery and spawn with asyncio.TaskGroup
Show how to mock Trio's nursery pattern using Python's asyncio.TaskGroup to simulate task spawning and completion.
import asyncio
class MockSpawner:
async def spawn(self, nursery):
print("Spawning mock task...")
await asyncio.sleep(1)
print("Mock task completed")
async def open_nursery():
async with asyncio.TaskGroup() as nursery:
mock = MockSpawner()
nursery.create_task(mock.spawn…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 …asyncio Condition wait notify pattern in Python
Coordinate coroutines with asyncio.Condition: workers wait for notifications and the main task notifies one or all of them.
import asyncio
async def worker(condition, name):
async with condition:
print(f"{name} waiting...")
await condition.wait()
print(f"{name} notified!")
async def main():
condition = asyncio.Condition()
tasks = [asyncio.create_task(worker(condition, f"worker-{i}")) for i in range(3…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 …Implement Bulkhead Thread Pool Isolation in Python
Create isolated thread pools with a bulkhead pattern to protect different services from cascading failures.
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
class Bulkhead:
"""Simple bulkhead isolation: separate thread pools for different tasks."""
def __init__(self, max_workers):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.active = …How to Implement ETag Optimistic Concurrency in Python
Build a lightweight in-memory resource store that uses MD5 hash ETags to prevent lost updates via optimistic concurrency control.
import hashlib
import json
class ResourceStore:
def __init__(self):
self.data = {}
self.etags = {}
def get(self, resource_id):
if resource_id not in self.data:
return None, None
return self.data[resource_id], self.etags[resource_id]
def put(self, resource_id, …Implement If-Match Precondition Update in Python
A mock resource store that uses the If-Match header's ETag to guard updates, preventing overwrites from stale clients.
from dataclasses import dataclass
from typing import Optional
@dataclass
class Resource:
id: str
version: int = 1
data: str = ""
etag: str = "etag-1"
class MockResourceStore:
def __init__(self):
self.resources = {}
def update(self, resource_id: str, new_data: str, if_match: Optiona…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…How to implement Redlock distributed lock in Python
Simulate Redis Redlock multi-instance locking to show how a distributed lock is acquired only when a majority of instances agree.
import time
import random
import threading
from dataclasses import dataclass
@dataclass
class MockRedisLock:
"""Simple mock of a Redis lock instance."""
name: str
key: str
ttl: int
acquired: bool = False
expires_at: float = 0.0
def acquire(self, sleep_fn=time.sleep):
"""Try to ac…Mock Redis Distributed Lock in Python with SET NX EX
A minimal in-memory mock of Redis SET NX EX distributed lock semantics for testing concurrent code without a real Redis server.
import time
import threading
import uuid
from typing import Optional
class RedisLockMock:
"""A minimal mock of Redis SET NX EX distributed lock semantics."""
def __init__(self):
self._store = {} # key -> (value, expiry_epoch)
def acquire(self, key: str, token: str, ttl_seconds: int) -> bool:
…Python Redis WATCH optimistic lock mock
A MockRedis class that simulates WATCH/MULTI/EXEC transactions with optimistic locking to detect concurrent modifications before committing.
import time
import threading
class MockRedis:
def __init__(self):
self.data = {}
self.watched = {}
self.lock = threading.Lock()
def get(self, key):
return self.data.get(key)
def set(self, key, value):
self.data[key] = value
def watch(self, *keys):
wi…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 Propagate Context Variables with asyncio in Python
Use Python's ContextVar with asyncio to carry deadline information across concurrent tasks and propagate context automatically.
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.…Bulkhead Thread Pool per Service Mock in Python
Simulates a bulkhead pattern with per-service thread pools and semaphore-based rejection to isolate failures between dependent services.
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
class ServiceBulkhead:
def __init__(self, name, max_threads, max_queue):
self.name = name
self.executor = ThreadPoolExecutor(max_workers=max_threads)
self.semaphore = threading.Semaphore(max_thread…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.