Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
How to Build a Message Stream Queue in Python
A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.
from collections import deque
import time
class MessageStream:
def __init__(self, messages):
self._queue = deque(messages)
self._sent = []
def send_next(self):
if not self._queue:
return None
message = self._queue.popleft()
self._sent.append(message)
…How to Implement Publish-Subscribe Fanout with Multiple Subscribers in Python
Create a simple publish-subscribe system in Python that broadcasts messages to multiple subscriber callbacks for a given topic.
import time
class PubSub:
def __init__(self):
self.subscribers = {}
def subscribe(self, topic, callback):
if topic not in self.subscribers:
self.subscribers[topic] = []
self.subscribers[topic].append(callback)
def publish(self, topic, message):
if topic in sel…How to Implement a Priority Queue for Messages in Python
Build a message priority queue with heapq and dataclasses that pops messages by priority, using sequence numbers to keep insertion order.
import heapq
from dataclasses import dataclass, field
from typing import Any
@dataclass(order=True)
class Message:
priority: int
sequence: int = field(compare=False)
content: str = field(compare=False)
class PriorityQueue:
def __init__(self):
self._heap = []
def push(self, priority: int,…How to Implement an In-Memory Pub/Sub System in Python
This code implements a simple in-memory publish/subscribe system in Python, allowing topics, callbacks, and message broadcasting.
class PubSub:
def __init__(self):
self.topics = {}
def subscribe(self, topic, callback):
if topic not in self.topics:
self.topics[topic] = []
self.topics[topic].append(callback)
return lambda: self.unsubscribe(topic, callback)
def unsubscribe(self, topic, callb…How to Mock Kafka Topic Partitions with a Python dict of lists
Mocks a Kafka topic and its partitions using a defaultdict of lists to simulate message production, consumption, and per-partition counts.
from collections import defaultdict
class KafkaTopicPartitionMock:
"""A simple mock for Kafka topic-partition assignment using dict of lists."""
def __init__(self, topic):
self.topic = topic
self.partitions = defaultdict(list) # partition_id -> list of messages
def produce(self, message…How to Mock MQTT Topic Subscriptions with QoS in Python
Build a lightweight MQTT client mock that tracks topic subscriptions with QoS levels and simulates wildcard message delivery.
import time
from collections import defaultdict
class MockMQTTClient:
def __init__(self):
self.subscriptions = defaultdict(list)
self.messages = []
def subscribe(self, topic, qos=0):
self.subscriptions[topic].append(qos)
print(f"Subscribed to '{topic}' with QoS {qos}")
…How to Mock RabbitMQ Ack Nack Requeue in Python
A mock RabbitMQ channel and consumer that simulates ack, nack, and requeue handling for testing message processing logic without a broker.
import json
from collections import deque
class MockChannel:
def __init__(self):
self.acked = []
self.nacked = []
self.requeued = []
def basic_ack(self, delivery_tag):
self.acked.append(delivery_tag)
def basic_nack(self, delivery_tag, requeue=False):
self.nacked.…How to Partition and Order Kafka-Style Messages by Key in Python
Group messages with the same key into ordered buckets using hashing and a defaultdict, mimicking Kafka partition ordering.
from dataclasses import dataclass
from collections import defaultdict
@dataclass
class Message:
key: str
content: str
def partition_and_order(messages, num_partitions=3):
partitions = defaultdict(list)
for msg in messages:
partition_id = hash(msg.key) % num_partitions
partitions[parti…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…How to deduplicate messages by ID in Python
Track seen message IDs in a set to skip duplicate messages and store unique content in a dict, with exact output showing which messages were added or skipped.
import time
class MessageStore:
def __init__(self):
self.seen_ids = set()
self.messages = {}
def add(self, message_id, content, timestamp=None):
timestamp = timestamp or time.time()
if message_id in self.seen_ids:
return False
self.seen_ids.add(message_…Implement a FIFO Message Queue in Python with deque
This code implements a FIFO (first-in-first-out) message queue class using Python's collections.deque, providing enqueue, dequeue, peek, and size operations.
from collections import deque
class MessageQueue:
def __init__(self):
self.queue = deque()
def enqueue(self, message):
self.queue.append(message)
print(f"Enqueued: {message}")
def dequeue(self):
if self.is_empty():
print("Queue is empty, cannot dequeue.")
…Mock NATS queue group load balancing in Python
Simulates a NATS queue group where each message is delivered to exactly one subscriber using random selection with a lightweight mock.
import random
import time
from collections import defaultdict
class MockQueueGroup:
"""Mock a NATS queue group: each message is delivered to exactly one subscriber."""
def __init__(self, subscribers):
self.subscribers = subscribers
def publish(self, message):
receiver = random.choice(se…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:
…Using the retained message flag in MQTT with Python
This script subscribes to an MQTT topic and prints the retained flag for each received message, demonstrating how to distinguish retained messages from normal ones.
import paho.mqtt.client as mqtt
def on_connect(client, userdata, flags, rc):
print(f"Connected with result code {rc}")
# Subscribe to a topic and check retained flag
client.subscribe("test/retained")
print("Subscribed to test/retained")
def on_message(client, userdata, msg):
# msg.retain is the M…How to Deduplicate Messages in Python by ID
This code consumes a mock inbox of JSON messages and deduplicates them by message ID, keeping either the first or last occurrence.
import json
from collections import OrderedDict
mock_inbox = [
{"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
{"id": 2, "message": "world", "timestamp": "2024-01-01T10:01:00Z"},
{"id": 1, "message": "hello", "timestamp": "2024-01-01T10:00:00Z"},
{"id": 3, "message": "test", "times…How to Implement Message Visibility Timeout Renewal in Python
Simulate queue message visibility control with timeout renewal using a simple Python class that tracks received time and visibility state.
import time
import uuid
class Message:
def __init__(self, body, visibility_timeout=30):
self.body = body
self.visibility_timeout = visibility_timeout
self.receipt_handle = str(uuid.uuid4())
self.received_at = time.time()
self.deleted = False
def is_visible(self):
…How to Implement a Dead Letter Queue Replay in Python
A mock Dead Letter Queue that stores failed messages with retry attempts and replays them with a simple retry counter.
import json
from collections import deque
class DeadLetterQueue:
def __init__(self):
self.messages = deque()
def add_message(self, message_id, payload, attempts=3):
"""Add a message to the DLQ with retry metadata."""
self.messages.append({
"id": message_id,
…How to Do Structured JSON Line Logging in Python
Create a simple JSON-lines logger that writes one JSON object per line to stdout with timestamp, level, message, and custom context fields.
import json
import sys
from datetime import datetime
class JsonLineLogger:
def __init__(self, stream=sys.stdout):
self.stream = stream
def log(self, level, message, **context):
record = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": level,
"me…How to Do Structured JSON Logging in Python
Create a custom logging formatter that outputs each log entry as a single JSON line with timestamp, level, logger name, and message.
import json
import logging
from datetime import datetime
class JsonFormatter(logging.Formatter):
def format(self, record):
log_entry = {
"timestamp": datetime.utcnow().isoformat() + "Z",
"level": record.levelname,
"logger": record.name,
"message": record.ge…How to Parse Log Lines with Regex in Python
Extracts timestamp, log level, service name, and message from a log line using compiled regex named groups.
import re
LOG_PATTERN = re.compile(
r'^(?P<timestamp>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) '
r'\[(?P<level>\w+)\] '
r'\((?P<service>[^)]+)\) '
r'(?P<message>.*)$'
)
def parse_log_line(line: str) -> dict:
match = LOG_PATTERN.match(line)
if not match:
return {"error": "invalid log format…How to Redact Secrets from Log Messages in Python
Build a lightweight RedactingFormatter class that replaces sensitive tokens like passwords and API keys with [REDACTED] before log messages are printed.
class RedactingFormatter:
def __init__(self, secrets):
self.secrets = secrets
def redact(self, message):
for secret in self.secrets:
message = message.replace(secret, "[REDACTED]")
return message
def format(self, record):
message = record["message"]
ret…How to Deduplicate Events in Python with SHA256 Hashing
Build an event deduplicator that identifies duplicate inbox messages using SHA256 hashes and tracks duplicate counts per event type.
```python
import hashlib
import json
from collections import defaultdict
class EventDeduplicator:
def __init__(self):
self.seen_hashes = set()
self.duplicate_counts = defaultdict(int)
def process_event(self, event):
event_key = f"{event['event_id']}:{event['timestamp']}"
even…How to mock resource request limits in Python
A Python class that simulates CPU and memory limit checks for resource requests, returning clear acceptance or rejection messages.
class ResourceLimits:
def __init__(self, cpu_limit, memory_limit):
self.cpu_limit = cpu_limit
self.memory_limit = memory_limit
def check_request(self, cpu, memory):
if cpu > self.cpu_limit:
return "CPU limit exceeded: {cpu} > {limit}".format(cpu=cpu, limit=self.cpu_limit)
…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.