Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
Synchronize Threads with a Barrier in Python
Demonstrates using threading.Barrier to synchronize multiple threads at phase boundaries, ensuring all workers wait for each other before proceeding.
import threading
import time
from random import randint
def worker(barrier, worker_id):
for phase in range(3):
time.sleep(randint(1, 3))
print(f"Worker {worker_id} finished phase {phase} at {time.time():.2f}")
barrier.wait()
print(f"Worker {worker_id}: all phases complete")
if __name_…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…How to Compare Execution Speed Between Python Functions
Measure and compare the average execution time of multiple Python functions using a reusable benchmark helper with time.perf_counter.
import time
import random
def method_a(values):
"""Sort using built-in sorted."""
return sorted(values)
def method_b(values):
"""Sort using list's sort method."""
values_copy = values[:]
values_copy.sort()
return values_copy
def method_c(values):
"""Sort manually using bubble sort (slow,…How to Parametrize pytest Tests with Multiple Input Cases in Python
This code shows how to use pytest's @pytest.mark.parametrize decorator to run the same test function across multiple input-output combinations, checking that an add function behaves correctly for each case.
import pytest
def add(a, b):
return a + b
@pytest.mark.parametrize("a,b,expected", [
(1, 2, 3),
(5, 5, 10),
(-1, 1, 0),
(0, 0, 0),
(10, -3, 7),
])
def test_add(a, b, expected):
assert add(a, b) == expected
if __name__ == "__main__":
pytest.main([__file__, "-v"])How to Skip Slow Tests with pytest.mark in Python
Use pytest.mark.skip and custom marks like @pytest.mark.slow to skip or deselect slow tests during test runs.
import pytest
import time
def test_fast():
assert 1 + 1 == 2
@pytest.mark.skip(reason="slow test skipped by default")
def test_slow():
time.sleep(5)
assert True
@pytest.mark.slow
def test_marked_slow():
time.sleep(5)
assert True
if __name__ == "__main__":
pytest.main([__file__, "-v", "-…How to Use Mock Flip Mutation Testing in Python
Demonstrates how mutation testing tools flip Boolean literals (mock flip) in Python source to verify test suite effectiveness in catching logic changes.
import random
# In mutation testing, a "mock flip" intentionally changes a Boolean
# constant to False (or True) to see if the test suite catches it.
# This is a common "constant mutation" applied to a source file's literals.
def is_even(n: int) -> bool:
"""Return True if n is even. Contains a Boolean literal us…How to Use Union Type Hints in Python
This code demonstrates how to use Union type hints to specify that a parameter can accept multiple types (int, float, str) and handle them accordingly.
from typing import Union
def process_value(value: Union[int, float, str]) -> str:
if isinstance(value, (int, float)):
return f"Number: {value * 2}"
return f"String: {value.upper()}"
if __name__ == "__main__":
print(process_value(10))
print(process_value(3.14))
print(process_value("hello"))Mock datetime with time-machine in Python
Use the time-machine library to travel to a fixed datetime when running tests or scripts, mocking datetime.utcnow().
from time_machine import travel
from datetime import datetime
@travel("2020-01-01 10:30:00")
def check_date():
return datetime.utcnow()
if __name__ == "__main__":
print(check_date())Domain Driven Design Aggregate Root Example in Python
Model an Order as an aggregate root with invariants enforced through methods, demonstrating DDD principles in Python.
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Optional
from uuid import uuid4
class Money:
def __init__(self, amount: float, currency: str = "USD"):
self.amount = amount
self.currency = currency
def __add__(self, other: Money) -> Money:
…How to Build a Pipe and Filter Text Processing Chain in Python
A functional pipe-and-filter chain that transforms text through uppercase, whitespace normalization, number removal, stopword filtering, and file export.
import re
import sys
def pipe_filter_chain(stream):
def uppercase(text):
return text.upper()
def strip_whitespace(text):
return " ".join(text.split())
def remove_numbers(text):
return re.sub(r"\d+", "", text)
def remove_stopwords(text, stopwords={"the", "and", "of", "in"}):…Idempotent Consumer: Store Processed IDs in Python
Implement an idempotent consumer that persists processed message IDs to a JSON file, skipping duplicates on restart.
import json
from pathlib import Path
class IdempotentStore:
def __init__(self, storage_path: str = "processed_ids.json"):
self.storage_path = Path(storage_path)
self.processed_ids = self._load()
def _load(self) -> set:
if self.storage_path.exists():
with self.storage_path…Generate an OpenAPI Spec from Mock Routes in Python
This Python script generates an OpenAPI 3.0 specification from a simple mock routes dictionary, mapping each HTTP method to response examples.
import json
from pathlib import Path
def generate_openapi_spec(routes: dict, title: str = "Mock API", version: str = "1.0.0") -> dict:
paths = {}
for route, methods in routes.items():
path_item = {}
for method, response_data in methods.items():
method = method.lower()
…How to Parse Multipart Form Data in Python
Parse multipart/form-data uploads using the Python standard library's cgi module to extract both regular fields and file uploads.
import cgi
from io import BytesIO
def parse_multipart_form(headers, body_bytes):
content_type = headers.get("Content-Type", "")
content_length = int(headers.get("Content-Length", len(body_bytes)))
# Create a file-like object from bytes for cgi.FieldStorage
body_file = BytesIO(body_bytes)
…Dead Letter Queue Failed Messages List Mock in Python
Implements a simple in-memory dead letter queue to collect, list, and retry failed messages, with JSON serialization for inspection in streaming pipelines.
import json
from collections import deque
class Message:
def __init__(self, message_id, payload, attempts=0):
self.message_id = message_id
self.payload = payload
self.attempts = attempts
def __repr__(self):
return f"Message(id={self.message_id}, attempts={self.attempts})"
c…How to Build a Flow Control Credit Window in Python
A Python class that reserves, confirms, releases, and settles credit to limit message flow and prevent overload in streaming pipelines.
class CreditWindow:
def __init__(self, max_credit=1000):
self.max_credit = max_credit
self.used_credit = 0
self.pending_credit = 0
def try_reserve(self, amount):
available = self.max_credit - self.used_credit - self.pending_credit
if available >= amount:
…How to Build a Mock Change Data Capture Event Stream in Python
Generate a deterministic list of mock CDC events with event IDs, stream positions, payloads, and timestamps for testing streaming pipelines.
from itertools import count
from random import choice, randint, seed
from datetime import datetime, timedelta
seed(42) # Make output deterministic
event_types = ["INSERT", "UPDATE", "DELETE"]
table_names = ["users", "orders", "products", "payments"]
counter = count(1)
def mock_cdc_event(stream_index: int) -> dict:
…How to Encode and Decode Avro Data in Python (Roundtrip)
Serialize a Python dict to Avro binary bytes and decode it back using the fastavro-compatible avro library.
import io
import json
from avro.schema import parse
from avro.io import DatumWriter, DatumReader, BinaryEncoder, BinaryDecoder
def avro_roundtrip(schema_json, data):
schema = parse(json.dumps(schema_json))
bytes_writer = io.BytesIO()
encoder = BinaryEncoder(bytes_writer)
writer = DatumWriter(schema)
…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 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 a Kafka Producer Batch Send in Python
Simulate a Kafka producer in Python that sends batched JSON events with mock partitions and latency for testing streaming pipelines without a real broker.
import json
import random
import time
from datetime import datetime
class MockKafkaProducer:
def __init__(self, topic):
self.topic = topic
self.sent_messages = []
def send(self, value, key=None):
message = {
"topic": self.topic,
"key": key,
"value"…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_…How to mock RabbitMQ queue binding with routing keys in Python
A mock demonstration of binding a queue to an exchange with multiple routing keys in RabbitMQ using Python and pika, without a real broker connection.
import pika
import sys
def bind_queue_with_routing(channel, queue_name, exchange_name, routing_keys):
"""
Mock RabbitMQ queue binding with routing keys.
Prints the binding configuration instead of connecting to a real broker.
"""
for routing_key in routing_keys:
binding = {
"q…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…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…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.