Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Initialize Sentry SDK with a Mock DSN in Python
Initialize the Sentry SDK in Python with a mock DSN to test error tracking without sending real events, then verify the DSN configuration.
import sentry_sdk
# Initialize Sentry SDK with a mock DSN (no real events will be sent)
sentry_sdk.init(
dsn="https://mock-public@mock-host/mock-project",
traces_sample_rate=1.0,
environment="development",
)
# Capture a test message to confirm SDK is configured
sentry_sdk.capture_message("Test message fr…How to Mock anyio.run Backends (asyncio vs trio) in Python
Demonstrates how to mock anyio.run to verify backend selection (asyncio or trio) without actually running the event loop.
import anyio
from unittest.mock import Mock, patch
async def fetch_data():
await anyio.sleep(0.1)
return {"data": 42}
def run_with_backend(backend: str):
async def main():
result = await fetch_data()
print(f"[{backend}] Result: {result}")
anyio.run(main, backend=backend)
if __nam…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 Run an Async Main with asyncio.run in Python
Show the canonical entry point for an asyncio program: define an async main, then launch it with asyncio.run.
import asyncio
async def main():
print("Hello from async main")
await asyncio.sleep(0.1)
print("Done")
if __name__ == "__main__":
asyncio.run(main())How to Signal asyncio Workers to Stop with an Event in Python
Use an asyncio.Event to coordinate graceful shutdown of multiple concurrent worker tasks in Python.
import asyncio
import random
async def worker(name, stop_event):
while not stop_event.is_set():
await asyncio.sleep(random.uniform(0.1, 0.5))
print(f"Worker {name} processing...")
print(f"Worker {name} stopped.")
async def main():
stop_event = asyncio.Event()
workers = [asyncio.create…How to Use a Weakref Cache to Avoid Memory Leaks in Python
This code demonstrates building a value cache with weakref.WeakValueDictionary so objects can be garbage collected when no longer referenced, preventing memory leaks.
import weakref
import gc
class ExpensiveObject:
def __init__(self, name):
self.name = name
def __repr__(self):
return f"ExpensiveObject('{self.name}')"
class ObjectCache:
def __init__(self):
self._cache = weakref.WeakValueDictionary()
def get_or_create(self, name):
…How to Use threading.Lock to Synchronize a Counter in Python
Safely increment a shared counter across multiple threads using threading.Lock as a mutex to prevent race conditions.
import threading
counter = 0
lock = threading.Lock()
def increment():
global counter
for _ in range(100000):
with lock:
counter += 1
threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"Final counter valu…How to Use uvloop Faster Event Loop
Install uvloop at startup to replace asyncio's default event loop with a faster libuv-based one, with a graceful fallback when it's unavailable.
import asyncio
try:
import uvloop
uvloop.install()
USING_UVLOOP = True
except ImportError:
USING_UVLOOP = False
async def fetch_data(index):
await asyncio.sleep(0.01)
return f"data-{index}"
async def main():
tasks = [fetch_data(i) for i in range(10)]
results = await asyncio.gather(*…How to set a timeout with asyncio.wait_for in Python
Use asyncio.wait_for to bound an async function with a timeout, catching TimeoutError when it exceeds the limit.
import asyncio
async def slow_task():
await asyncio.sleep(3)
return "finished"
async def main():
try:
result = await asyncio.wait_for(slow_task(), timeout=1)
print(result)
except asyncio.TimeoutError:
print("Task timed out")
if __name__ == "__main__":
asyncio.run(main())Run Background Tasks with asyncio.create_task in Python
Create background tasks in an asyncio event loop with asyncio.create_task and run them concurrently using asyncio.gather.
import asyncio
import time
async def background_worker(name, duration):
"""Simulates a long-running background task."""
print(f"{name} started at t={time.monotonic():.1f}")
await asyncio.sleep(duration)
print(f"{name} finished at t={time.monotonic():.1f}")
async def main():
print(f"Main starting …asyncio sleep cooperative scheduling demo in Python
This demo shows how asyncio.sleep yields control between concurrent tasks, letting multiple workers interleave their ticks.
import asyncio
async def worker(name, delay):
for i in range(3):
print(f"{name}: tick {i}")
await asyncio.sleep(delay)
return f"{name} done"
async def main():
tasks = [
asyncio.create_task(worker("A", 0.1)),
asyncio.create_task(worker("B", 0.2)),
asyncio.create_tas…Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States
Implement a circuit breaker with closed, open, and half-open states to prevent repeated calls to failing services and allow recovery after a timeout.
class CircuitBreaker:
def __init__(self, failure_threshold=3, timeout_seconds=5):
self.failure_threshold = failure_threshold
self.timeout_seconds = timeout_seconds
self.state = "closed"
self.failure_count = 0
self.last_failure_time = None
def record_success(self):
…How to Build an Append-Only Event Store in Python
Implement a simple append-only event store class that stores events in a list and supports retrieval by index range.
class EventStore:
def __init__(self):
self._events = []
def append(self, event):
"""Append an event to the store."""
self._events.append(event)
def get_events(self, start=0, end=None):
"""Return events from start index to end (exclusive)."""
return self._events[sta…How to Implement a Simple Event Bus in Python
Create a publish-subscribe event bus using dataclasses and defaultdict to decouple event producers from consumers.
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Set
@dataclass
class EventBus:
_subscribers: Dict[str, List[Callable]] = field(
default_factory=lambda: defaultdict(list)
)
def subscribe(self, event_type: str, handler: Callable…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 …Inbox pattern consumer dedupe mock in Python
Implements a mock inbox consumer that deduplicates incoming messages by ID, with automatic eviction of old seen IDs to prevent unbounded memory growth.
import json
from collections import deque
from dataclasses import dataclass, field
from hashlib import sha256
from typing import Any
@dataclass
class InboxConsumer:
max_seen: int = 1000
seen_ids: set = field(default_factory=set)
seen_history: deque = field(default_factory=deque)
def _mark_seen(self,…Outbox pattern reliable publish in Python with SQLite
Implements a transactional outbox with SQLite, ensuring reliable message publishing by storing events in the same DB transaction as business changes.
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
class Outbox:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS outbox (
id INTEGER PRIMARY KEY AUTO…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, …How to mock Server-Sent Events (SSE) in Python
A minimal HTTP server that streams Server-Sent Events to clients, perfect for testing and development.
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time
MESSAGES = iter([
"data: Hello world\n\n",
"data: Second message\n\n",
"event: custom\n",
"data: Custom event payload\n\n",
"data: Final message\n\n"
])
class SSEHandler(BaseHTTPRequestHandler):
def do_GET…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…Event Envelope with Schema Version Field in Python
Build a typed event envelope dataclass with an explicit schema version field for mock streaming scenarios.
from dataclasses import dataclass, field
from datetime import datetime
import uuid
@dataclass
class Event:
event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
event_type: str = "user.created"
version: str = "1.0.0"
created_at: str = field(default_factory=lambda: datetime.utcnow().isoform…Event sourcing append store replay in Python
A simple in-memory event store that appends events per aggregate and replays them on demand.
import json
from collections import defaultdict
class EventStore:
def __init__(self):
self._events = defaultdict(list)
def append(self, aggregate_id, event_type, data):
event = {"type": event_type, "data": data}
self._events[aggregate_id].append(event)
def replay(self, aggregate…Exactly Once Idempotent Consumer Store in Python
A mock key-value store that guarantees exactly-once processing by rejecting duplicate message keys in a message or event stream.
from collections import defaultdict
class ExactlyOnceStore:
def __init__(self):
self.processed = defaultdict(set)
self.data = {}
def consume(self, key, value):
if key in self.data:
return False
self.data[key] = value
return True
def get_processed_count…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.