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…Mock Redis Streams XADD and XREAD in Python
A pure-Python mock of Redis streams that implements basic XADD, XREAD, and XLEN behavior for local testing without a real Redis server.
import redis
import time
import threading
class MockRedisStreams:
def __init__(self):
self.streams = {}
def xadd(self, stream_name, fields):
if stream_name not in self.streams:
self.streams[stream_name] = []
entry_id = f"{time.time_ns()}-{len(self.streams[stream_name])}"
…Redis Pub/Sub Channel Subscribe Mock in Python
A lightweight in-memory mock of Redis pub/sub that lets you subscribe to channels, publish messages, and verify handler behavior in tests without a real Redis server.
class MockRedisPubSub:
def __init__(self):
self.channels = {}
def subscribe(self, channel):
if channel not in self.channels:
self.channels[channel] = []
return self.channels[channel]
def publish(self, channel, message):
if channel in self.channels:
…Cache Data in Redis with Python
A beginner-friendly Redis cache helper that stores JSON strings with a TTL and retrieves them with the redis-py client.
import redis
class DataCache:
def __init__(self, host="localhost", port=6379, db=0):
self.client = redis.Redis(host=host, port=port, db=db, decode_responses=True)
def cache_data(self, key, value, ttl=60):
self.client.setex(key, ttl, value)
def get_cached_data(self, key):
return …How to Build a Redis Leaderboard with ZREVRANGE in Python
Build a sorted leaderboard by storing player scores as a Redis sorted set and reading the top scores with ZREVRANGE in Python.
import redis
import random
# Connect to local Redis (ensure Redis is running on localhost:6379)
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
# Clear any existing test data
r.delete("game_scores")
# Simulate player scores
players = ["alice", "bob", "charlie", "dave", "eve"]
for player in…How to Cache Data in Redis with Python
Build a simple Redis cache wrapper that stores and retrieves JSON data with automatic TTL and serialization.
import redis
import json
import time
class Cache:
def __init__(self, host="localhost", port=6379, db=0, default_ttl=60):
self.client = redis.Redis(host=host, port=port, db=db)
self.default_ttl = default_ttl
def get(self, key):
value = self.client.get(key)
if value is None:
…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.
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…How to Cache Function Results with Redis in Python
A RedisCache helper class caches function results using a decorator, with JSON serialization and TTL-based expiry.
import redis
import json
from functools import wraps
class RedisCache:
def __init__(self, host='localhost', port=6379, db=0, ttl=60):
self.client = redis.Redis(host=host, port=port, db=db, decode_responses=True)
self.ttl = ttl
def cached(self, key_prefix):
def decorator(func):
…How to Implement a Redis-Like Cache Dictionary in Python
Build a RedisMockDict class that mimics basic Redis key-value operations with TTL support, expiry cleanup, and standard dict-like methods.
from collections import OrderedDict
import time
class RedisMockDict:
def __init__(self, ttl=None):
self._data = OrderedDict()
self._ttl = ttl # default TTL in seconds, None = no expiry
self._expiry = {}
def set(self, key, value, ttl=None):
"""Set a key-value pair with optiona…How to Iterate Redis Keys with SCAN in Python
Iterate all Redis keys matching a pattern using the SCAN command with a mock client to simulate pagination.
import redis
def scan_keys(client, pattern="*", count=10):
keys = []
cursor = 0
while True:
cursor, batch = client.scan(cursor=cursor, match=pattern, count=count)
keys.extend(batch)
if cursor == 0:
break
return keys
if __name__ == "__main__":
# Mock Redis clien…How to Mock Redis EXPIRE, TTL, and PERSIST in Python
A lightweight in-memory MockRedis class that simulates Redis key expiration, TTL, and persist behavior for tests and local development.
import time
class MockRedis:
def __init__(self):
self._store = {}
self._expiry = {}
def set(self, key, value):
self._store[key] = value
self._expiry.pop(key, None)
return True
def expire(self, key, ttl_seconds):
if key not in self._store:
retur…How to Mock Redis Pipeline Batch Commands in Python
Create a lightweight MockRedis class that simulates Redis pipeline batching with SET, GET, and DELETE operations for testing without a live server.
import redis
import time
class MockRedis:
def __init__(self):
self.data = {}
def pipeline(self):
return MockPipeline(self)
def execute(self, commands):
results = []
for cmd in commands:
op, args = cmd[0], cmd[1:]
if op == "SET":
se…How to Mock Redis Pub/Sub in Python
Test Redis pub/sub logic without a live server using an in-memory fake that queues published messages per channel.
import redis
import time
import threading
class MockRedisPubSub:
def __init__(self):
self.channels = {}
def publish(self, channel, message):
if channel not in self.channels:
return 0
for subscriber in self.channels[channel]:
subscriber.put(message)
ret…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 Mock a Redis Session Store Cookie SID in Python
Mock a Redis-backed session store with a cookie-based session ID (SID) in Python, including the create, read, and delete operations.
import redis
import uuid
import time
class RedisSessionStore:
def __init__(self, host="localhost", port=6379, db=0, prefix="session:"):
self.client = redis.Redis(host=host, port=port, db=db)
self.prefix = prefix
def create_session(self, timeout_seconds=3600):
session_id = uuid.uuid4(…How to Mock a Redis Transaction with MULTI/EXEC in Python
A minimal in-memory mock of Redis MULTI/EXEC transactions that queues commands and applies them atomically on EXEC.
class RedisTransactionMock:
def __init__(self):
self.data = {}
self.queue = []
self.in_transaction = False
def multi(self):
self.in_transaction = True
self.queue = []
return "OK"
def set(self, key, value):
if self.in_transaction:
self.qu…How to Serialize Cache Values with JSON and Pickle in Python
Serialize cache values using JSON for simple types or pickle for arbitrary objects, with robust error handling for unsupported types like mocks.
import json
import pickle
from unittest.mock import Mock
def serialize(value, method="json"):
"""Serialize a cache value using JSON or pickle with type checking."""
if method == "json":
try:
return json.dumps(value).encode("utf-8")
except TypeError as e:
raise ValueErro…How to Use Redis HSET and HGET in Python
This code demonstrates how to store and retrieve hash data in Redis using Python's redis library with HSET, HGET, HGETALL, and HDEL commands.
import redis
# Connect to Redis (adjust host/port as needed)
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# Clear any existing data for demonstration
r.delete('user:1')
# HSET - Store a hash
r.hset('user:1', mapping={'name': 'Alice', 'age': 30, 'city': 'New York'})
# HGET - Retrieve a …How to Use Redis ZADD and ZRANGE in Python
Add members to a Redis sorted set with ZADD and retrieve them in score order with ZRANGE in Python.
import redis
client = redis.Redis(host='localhost', port=6379, db=0)
client.delete('scores')
members = {'alice': 30, 'bob': 20, 'carol': 50}
for name, score in members.items():
client.zadd('scores', {name: score})
result = client.zrange('scores', 0, -1)
print(result)How to Use Redis as a Cache in Python
A beginner-friendly RedisCache helper that stores, retrieves, and deletes JSON values with automatic TTL expiration using the redis-py client.
import json
import time
import redis
class RedisCache:
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 set(self, key, value, ttl=None):
"""Store a v…How to Validate and Cache Data with Redis in Python
A beginner-friendly helper that validates email, phone, and age data and caches validated entries in Redis for 5 minutes.
import redis
import json
from functools import wraps
class DataValidator:
def __init__(self, host="localhost", port=6379, db=0):
self.cache = redis.Redis(host=host, port=port, db=db)
self.validators = {
"email": lambda v: "@" in v and "." in v.split("@")[-1],
"phone": lambd…How to cache filtered data in Redis with Python
This code caches filtered list results in Redis using an MD5 hash key, returning cached results when available.
import redis
import json
import hashlib
import time
cache = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def filter_data(data, predicate_key, predicate_value):
"""Filter a list of dicts by key-value pair, with Redis caching."""
cache_key = hashlib.md5(
f"{predicate_key}:{pred…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…How to mock Redis geospatial commands (GEOADD) in Python
Implement a lightweight Python mock of Redis geospatial commands (GEOADD, GEODIST, GEOSEARCH) using the Haversine formula for testing without a Redis server.
import math
import heapq
class MockRedisGeo:
def __init__(self):
self.members = {}
def geoadd(self, key, longitude, latitude, member):
if key not in self.members:
self.members[key] = {}
self.members[key][member] = (longitude, latitude)
def geodist(self, key, member1,…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.