Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Read Redis Streams with XREADGROUP in Python
Read new messages from a Redis stream using a consumer group with XREADGROUP, handling JSON payloads and group creation.
import redis
import json
def read_group_messages(stream_key, group_name, consumer_name, count=10):
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
try:
r.xgroup_create(stream_key, group_name, id="0", mkstream=True)
except redis.exceptions.ResponseError:
pass
messag…How to Stream Join Windowed Mock Topics in Python
Simulates two message topics and joins their events when timestamps fall within a sliding time window using Python generators and deques.
import itertools
import random
import time
from collections import deque
from dataclasses import dataclass, field
@dataclass
class Event:
key: str
value: int
timestamp: float = field(default_factory=time.time)
def generate_topic(prefix, keys, start_time):
while True:
yield Event(
…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.
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…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.
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…Mock Watermark Late Event Side Output in Python
Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.
from datetime import datetime, timedelta
from typing import List, Tuple
def watermark_mock(
events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
"""Simulate watermarking: events arriving on time vs. late by ch…How to Build a Bloom Filter to Reduce Cache Misses in Python
Implement a probabilistic Bloom filter in Python that lets a cache quickly determine which keys are definitely not present, reducing expensive source lookups on cache misses.
import hashlib
import random
class BloomFilter:
def __init__(self, size=100, num_hashes=3):
self.size = size
self.num_hashes = num_hashes
self.bit_array = [0] * size
def _hashes(self, item):
result = []
for i in range(self.num_hashes):
hash_value = int(hash…How to Implement a Negative Cache with TTL in Python
This code provides a TTL mock cache that stores negative results (cache misses) for a short time to reduce repeated lookups of missing keys.
from time import time, sleep
class TTLMockCache:
def __init__(self, ttl_seconds=5):
self.ttl = ttl_seconds
self.store = {}
self.negative_cache = {}
def get(self, key):
now = time()
if key in self.store:
value, expires_at = self.store[key]
if exp…How to Mock Cache Tag Invalidation in Python
Use unittest.mock.patch with wraps to verify tagged cache entries are invalidated correctly.
import unittest
from unittest.mock import patch
def get_cached_data(cache, key):
"""Return data from cache if present and valid, else None."""
if cache.get(key, {}).get("valid", False):
return cache[key]["data"]
return None
def invalidate_tag_mock(cache, tag):
"""Invalidate all cache entries …How to Mock Redis Streams Consumer Groups in Python
Simulate Redis Streams producer and consumer group behavior in Python using a standalone mock class for testing and development.
import time
import json
from collections import defaultdict
class RedisStreamMock:
def __init__(self):
self.streams = defaultdict(list)
self.consumer_groups = defaultdict(dict)
self.pending_entries = defaultdict(list)
def xadd(self, stream, fields):
entry_id = f"{time.time_ns(…How to use Redis MGET MSET pipeline in Python
Store multiple keys atomically and read them efficiently with Redis MSET/MGET, then batch commands with a pipeline to cut round trips.
import redis # v4.x+ required
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
# Sample data to store
r.flushdb()
data = {"name": "Alice", "age": "30", "city": "Berlin"}
# MSET: store multiple key-value pairs in one command
r.mset(data)
# MGET: fetch multiple keys in one round trip
keys =…Redis-inspired sliding window rate limiter in Python
A pure-Python sliding window rate limiter using a deque of timestamps, mock-ready for Redis-backed production limits.
import time
from collections import deque
class SlidingWindowRateLimiter:
def __init__(self, max_requests: int, window_seconds: int) -> None:
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: dict[str, deque] = {}
def is_allowed(self, client_id: str…Refresh Proactive TTL Renewal in Python
This snippet implements a proactive TTL renewal pattern that refreshes a cache expiration before it lapses, using a mock counter to track renewals.
import time
from datetime import datetime, timezone
class TTLRenewer:
def __init__(self, ttl_seconds=10, renew_at=0.5):
self.ttl = ttl_seconds
self.last_renewed = time.time()
self.renew_threshold = ttl_seconds * renew_at
self.renewals = 0
def check_and_renew(self):
if …Simple Redis Cache Helper in Python
Build a minimal Redis-backed cache with TTL, JSON serialization, and automated fetching to speed up repeated expensive lookups.
import time
import redis
import json
class SimpleCache:
def __init__(self, host="localhost", port=6379, db=0, default_ttl=60):
self.client = redis.Redis(host=host, port=port, db=db, decode_responses=True)
self.default_ttl = default_ttl
def get(self, key):
value = self.client.get(key)…Build a Rate Limiter Decorator in Python
This code defines a reusable rate limiter decorator that caps function calls within a sliding time window using a deque and monotonic time.
import time
from collections import deque
def rate_limiter(max_calls: int, period: float):
calls = deque()
def decorator(func):
def wrapper(*args, **kwargs):
now = time.monotonic()
while calls and now - calls[0] >= period:
calls.popleft()
if len(ca…Health Check Mark Unhealthy Stop Traffic Mock in Python
Simulates a health check with a 20% failure rate and automatically stops traffic when the service is unhealthy.
import time
import random
class HealthCheck:
def __init__(self):
self.is_healthy = True
self.stop_traffic = False
def check_health(self):
# Simulate health check with random failure rate (20% chance unhealthy)
self.is_healthy = random.random() > 0.2
return self.is_heal…How to Build a Rate Limiter in Python
Implements a simple sliding-window rate limiter that caps the number of calls per period, used to throttle processing of a data list.
import time
class RateLimiter:
def __init__(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.timestamps = []
def allow(self):
now = time.time()
self.timestamps = [t for t in self.timestamps if now - t < self.period]
if len(self.tim…How to Cap Retry Attempts in Python with a Decorator
Build a reusable retry decorator that caps attempts, adds delays, and lets flaky services fail fast instead of hanging.
import random
from functools import wraps
from time import sleep
def retry(max_attempts, delay=0.1):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kw…How to Implement a Rate Limiter in Python
A beginner-friendly Python class that tracks call timestamps with a deque to allow or block calls based on a max rate per time period.
import time
from collections import deque
class RateLimiter:
"""Simple rate limiter for beginners."""
def __init__(self, max_calls: int, period_seconds: float):
self.max_calls = max_calls
self.period = period_seconds
self.calls = deque()
def allow(self) -> bool:
"""Retur…How to Implement a Sliding Window Log Rate Limiter in Python
Implements a sliding window log rate limiter in Python using a deque of timestamps to enforce a maximum request count within a rolling time window.
from collections import deque
from datetime import datetime, timedelta
from time import sleep
class SlidingWindowLog:
def __init__(self, window_seconds: int, max_requests: int):
self.window_seconds = window_seconds
self.max_requests = max_requests
self.timestamps = deque()
def allow_…How to implement rate limiting in Python
Build a simple sliding-window rate limiter in Python that enforces a max number of calls per time period and formats data with timestamps.
import time
class RateLimiter:
def __init__(self, max_calls, period):
self.max_calls = max_calls
self.period = period
self.calls = []
def allow(self):
now = time.time()
# Remove calls older than the period window
self.calls = [t for t in self.calls if now -…How to implement rate limiting per API key in Python
A simple sliding-window rate limiter that tracks request timestamps per API key and rejects requests exceeding the configured limit.
import time
API_RATE_LIMITS = {"api_key_1": 5, "api_key_2": 3} # max requests per window
WINDOW_SECONDS = 10
class RateLimiter:
def __init__(self, limits, window):
self.limits = limits
self.window = window
self.requests = {key: [] for key in limits}
def allow(self, api_key):
…Rate Limit per User ID in Python with a Dict Mock
Implements a simple sliding window rate limiter using a defaultdict of timestamps per user ID, blocking requests that exceed a max count within a time window.
import time
from collections import defaultdict
class RateLimiter:
def __init__(self, max_requests, window_seconds):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.user_timestamps = defaultdict(list)
def allow_request(self, user_id):
now = time.tim…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.
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:
…Saga Compensating Transaction Mock in Python
Simulates a distributed transaction using a saga pattern with compensating actions that roll back steps on failure.
import random
import time
class OrderService:
def __init__(self):
self.orders = {}
def create_order(self, order_id):
print(f"[Order] Creating order {order_id}...")
time.sleep(0.1)
if random.random() < 0.3: # 30% chance of failure
raise RuntimeError(f"Order {order…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.