Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
How to route late-arriving data to a side output in Python
Separate late-arriving events from a streaming data batch into a dead-letter side output list using a timestamp threshold.
from collections import defaultdict
def late_arriving_side_output(events, late_threshold_ts):
"""
Mock a streaming pipeline that separates late-arriving data events
into a side output list (e.g., for dead-letter analysis).
events: list of (timestamp, data) tuples, timestamps as ints.
late_thresho…Rollback dataset to previous snapshot pointer in Python
A SnapshotManager class stores timestamped data snapshots and rolls back to the most recent snapshot at or before a target time.
from datetime import datetime, timedelta
class SnapshotManager:
def __init__(self):
self.snapshots = {} # timestamp -> data
self.current_pointer = None
def create_snapshot(self, data):
timestamp = datetime.now()
self.snapshots[timestamp] = data
self.current_pointer =…Trigger a Pipeline When a New File Appears in a Directory
Poll a directory every 0.5 seconds and return the name of the first new file that appears, or None after a timeout.
import time
from pathlib import Path
def watch_for_file(directory: str, interval: float = 0.5, timeout: float = 10.0) -> str | None:
"""Poll a directory and trigger when a new file appears."""
watch_dir = Path(directory)
watch_dir.mkdir(exist_ok=True)
known_files = set(watch_dir.iterdir())
s…How to Mirror a Bare Git Repository Backup in Python
Run a git clone --bare subprocess to create a timestamped bare-repo backup folder with error handling.
import subprocess
import shlex
from pathlib import Path
from datetime import datetime
def mirror_bare_repo(source_url: str, backup_dir: str) -> str:
"""Mirror a bare git repository to a timestamped backup folder."""
backup_path = Path(backup_dir)
backup_path.mkdir(parents=True, exist_ok=True)
timest…Generate Mock CloudFormation Stack Events in Python
Generate a list of mock AWS CloudFormation stack events with random resources, statuses, and timestamps, and print them as JSON.
import json
import random
from datetime import datetime, timedelta
def generate_mock_stack_events(stack_name="MyTestStack", num_events=10):
"""Generate a list of mock CloudFormation stack events."""
resources = [
("AWS::S3::Bucket", "MyBucket"),
("AWS::EC2::Instance", "MyInstance"),
("…How to Create a JSON Data Helper in Python
A beginner-friendly DataHelper class that safely reads and writes JSON files with timestamps to a local data directory.
from datetime import datetime
from pathlib import Path
import json
class DataHelper:
"""Simple helper for reading/writing JSON files safely."""
def __init__(self, base_dir="data"):
self.base_dir = Path(base_dir)
self.base_dir.mkdir(exist_ok=True)
def save(self, filename, data):
…How to Create a Mock STS AssumeRole Credentials Dict in Python
Build a realistic AWS STS AssumeRole response dict with temporary credentials, expiry time, and assumed role ARN for local testing.
import json
from datetime import datetime, timedelta, timezone
def mock_sts_credentials(role_arn, session_name, duration=3600):
now = datetime.now(timezone.utc)
expiration = now + timedelta(seconds=duration)
credentials = {
"Credentials": {
"AccessKeyId": "ASIAEXAMPLEACCESSKEY",
…Mock ECS Task Run Stop Status Dict in Python
Build a mock ECS task status dictionary with RUNNING/STOPPED states using the standard library.
from datetime import datetime, timezone
def mock_ecs_task_status(task_id: str, state: str = "RUNNING") -> dict:
"""Return a mock ECS task status dictionary."""
return {
"taskArn": f"arn:aws:ecs:us-east-1:123456789012:task/cluster/{task_id}",
"taskDefinition": "arn:aws:ecs:us-east-1:1234567890…Mock GCP Secret Manager access version in Python
A minimal mock of GCP Secret Manager that stores secret versions, retrieves payloads by version, and logs access timestamps.
import json
import time
from datetime import datetime, timezone
class MockSecretManager:
"""Minimal mock of GCP Secret Manager access/version behavior."""
def __init__(self):
self._secrets = {}
self._access_log = []
def create_secret(self, secret_id: str, payload: str) -> dict:
…Data Conversion Helper Functions in Python
A set of beginner-friendly helper functions to convert between JSON strings and Python data, parse dates, and read/write files using pathlib.
from datetime import datetime
from pathlib import Path
import json
def to_json(data, indent=2):
"""Convert Python data to pretty-printed JSON string."""
return json.dumps(data, indent=indent, default=str)
def from_json(json_string):
"""Parse JSON string back into Python data."""
return json.loads(jso…How to Format Data with Python's datetime and JSON Helpers
A beginner-friendly set of helper functions to format dates and safely read/write JSON files in Python.
from datetime import datetime
from pathlib import Path
import json
def format_today(pattern: str = "%Y-%m-%d") -> str:
"""Return today's date formatted with the given pattern."""
return datetime.now().strftime(pattern)
def load_json(file_path: str) -> dict:
"""Read and parse a JSON file safely."""
…How to Generate a Mock Rollbar Error Report in Python
Create a realistic fake Rollbar error report with random timestamps, levels, messages, and counts for testing and demos.
import json
import random
import time
from datetime import datetime, timedelta
def mock_rollbar_report(n_errors=5):
messages = [
"TypeError: unsupported operand type(s) for +: 'int' and 'str'",
"KeyError: 'user_id'",
"ValueError: invalid literal for int() with base 10: 'abc'",
"At…How to Type Check a Mock with pyright in Python
Shows how pyright validates a mock function against a TypedDict and Callable signature before runtime.
from typing import TypedDict, Callable
class User(TypedDict):
id: int
name: str
def get_user_name(user_id: int, get_user: Callable[[int], User]) -> str:
user = get_user(user_id)
return user["name"]
def mock_get_user(user_id: int) -> User:
return {"id": user_id, "name": f"User {user_id}"}
if…How to Time Code Performance with timeit in Python
Benchmark two implementations of the same logic using Python's timeit module and compare their execution speeds.
import timeit
# Implementation 1: Using a list comprehension
def list_comprehension_squares(n):
return [i ** 2 for i in range(n)]
# Implementation 2: Using a for loop with append
def loop_squares(n):
result = []
for i in range(n):
result.append(i ** 2)
return result
if __name__ == "__main__"…How to Use ThreadPoolExecutor and ProcessPoolExecutor in Python
Compares ThreadPoolExecutor and ProcessPoolExecutor by running CPU-bound and I/O-tolerant tasks over a large list, printing elapsed times and first results.
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import math
numbers = list(range(1, 1000001))
def compute_square(n):
return n * n
def compute_sqrt(n):
return math.sqrt(n)
def run_executor(executor, func, data):
start = time.perf_counter()
results = list(executo…How to Use bisect.insort in Python to Maintain a Sorted List
Insert items into an already sorted list using Python's bisect.insort to keep it sorted efficiently in O(n) time.
import bisect
def maintain_sorted_list():
data = [3, 1, 4, 1, 5, 9, 2, 6]
sorted_list = []
for num in data:
bisect.insort(sorted_list, num)
print("Original data:", data)
print("Sorted list maintained with insort:", sorted_list)
# Insert new values to maintain sorted orde…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())How to use ThreadPoolExecutor for concurrent tasks in Python
Run blocking functions in parallel with ThreadPoolExecutor and as_completed, cutting total runtime from 5 sequential sleeps to about 1 second.
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def fetch_data(item):
"""Simulate a slow operation with a fixed delay."""
time.sleep(0.2)
return item * 2
def main():
items = [1, 2, 3, 4, 5]
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=3) as ex…How to freeze time in Python tests with freezegun
Use the freezegun decorator to freeze datetime.now() at a fixed timestamp so tests that depend on current time run deterministically.
from datetime import datetime
from freezegun import freeze_time
@freeze_time("2024-01-15 12:30:00")
def test_frozen_time():
now = datetime.now()
return now
if __name__ == "__main__":
result = test_frozen_time()
print(result)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())Mock datetime.now to freeze time in Python
Use unittest.mock.patch to replace datetime.now with a fixed value so your code always sees the same time during tests.
from datetime import datetime
from unittest.mock import patch
def current_message():
now = datetime.now()
return f"Current time: {now:%Y-%m-%d %H:%M:%S}"
if __name__ == "__main__":
with patch("__main__.datetime") as mock_dt:
mock_dt.now.return_value = datetime(2024, 3, 15, 10, 30, 0)
prin…How to Mock a Metrics Decorator in Python with unittest.mock
This code demonstrates a timing decorator that wraps a function to measure execution time and prints the duration, with a unit test using unittest.mock to patch the print function and assert it was called.
import time
from functools import wraps
from unittest.mock import patch
def add_metrics(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} took {elapsed:.6f}s…How to Poll an Operation Status Endpoint in Python
Mock a polling endpoint in Python that simulates checking an async operation's status until it completes or times out.
import time
import random
def poll_status(url: str, timeout: float = 5.0) -> dict:
"""Mock a polling endpoint that eventually returns a completed status."""
start = time.time()
while time.time() - start < timeout:
# Simulate delayed response
time.sleep(0.2)
# 80% chance to report …Build a Streaming Messaging Helper in Python
Create a simple message stream class that stores recent messages, sends user messages, and retrieves history or latest messages with timestamps.
from collections import deque
from dataclasses import dataclass
from datetime import datetime
import time
@dataclass
class Message:
user: str
text: str
timestamp: str = ""
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().strftime("%H:%M:%S")
class…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.