Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
Fan Out Records to Multiple Sinks in Python
Distribute the same records across multiple target sinks (database, API, queue, etc.) using a defaultdict-based fan-out pattern.
import json
from collections import defaultdict
SINKS = ["database", "api", "message_queue", "data_lake", "monitoring"]
def fan_out(records, *sinks):
dist = defaultdict(list)
for record in records:
for sink in sinks:
dist[sink].append(record)
return dict(dist)
if __name__ == "__main_…How to Implement a Sliding Window Average in Python
Compute the average of the most recent N values in a stream using a bounded deque, efficiently updating the total as new values arrive.
from collections import deque
class SlidingWindowAverage:
def __init__(self, window_size):
self.window_size = window_size
self.window = deque(maxlen=window_size)
self.total = 0
def add(self, value):
if len(self.window) == self.window_size:
self.total -= self.windo…How to Validate Data in a Python Pipeline
A helper module to validate common record types — email, positive integer, and non-empty string list — before processing data in a pipeline.
from typing import Any, Iterable
def is_valid_email(email: str) -> bool:
"""Basic email check: one '@', no spaces, dot after '@'."""
if "@" not in email or " " in email:
return False
local, _, domain = email.partition("@")
return bool(local) and "." in domain
def is_positive_int(value: Any)…Pipeline stage compose functions left to right in Python
Compose multiple functions into a left-to-right pipeline so each stage receives the output of the previous one.
def compose(*funcs):
"""Compose functions left to right: compose(f, g, h)(x) == h(g(f(x)))"""
def composed(arg):
result = arg
for func in funcs:
result = func(result)
return result
return composed
if __name__ == "__main__":
def add_one(x):
return x + 1
…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 =…Amend Last Commit Message in Python
This script uses subprocess to run `git commit --amend` and update the most recent commit's message in your repository.
import subprocess
import sys
def amend_last_commit_message(new_message: str) -> None:
"""Change the message of the most recent commit."""
result = subprocess.run(
["git", "commit", "--amend", "-m", new_message],
capture_output=True,
text=True,
check=False,
)
if result.…How to Filter Git History to Remove Secret File Entries in Python
A pure-Python mock that filters a repository's history to drop any commit that touched a secret file, so you can plan a cleanup before rewriting Git history.
from pathlib import Path
import json
def filter_history(history, secret_path):
"""Remove entries that touch the secret file."""
return [entry for entry in history if secret_path not in entry["files"]]
if __name__ == "__main__":
repo_history = [
{"commit": "a1b2c3", "message": "Add app", "files": …How to Generate Git LFS Extension Patterns in Python
This script builds mock Git LFS file patterns for common geospatial extensions and filters them based on compression suffixes.
import itertools
import re
LFS_EXTENSIONS = {".csv", ".geojson", ".tif", ".shp", ".gpkg"}
def build_mock_lfs_pattern(base_name="data_usgs_lidar"):
patterns = []
for ext in sorted(LFS_EXTENSIONS):
for variant in (("", ".lz4"), (".compressed",), (".b", ".a"), ("_v1", ".zip")):
full_pattern …How to List Changed Files in the Last Git Commit with Python
Runs `git diff --name-only HEAD~1 HEAD` via subprocess to list the names of files changed in the most recent commit.
import subprocess
def list_changed_files():
result = subprocess.run(
["git", "diff", "--name-only", "HEAD~1", "HEAD"],
capture_output=True,
text=True,
check=True
)
files = result.stdout.strip().splitlines()
return files
if __name__ == "__main__":
changed = list_cha…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…How to Push Git Tags to a Remote with Python
Push specified git tags (or all tags) to a remote repository using Python's subprocess module with error handling.
import subprocess
import sys
def push_tags_to_remote(remote: str = "origin", tags: list[str] | None = None) -> None:
"""
Push git tags to a remote repository.
If no tags are given, push all local tags.
"""
if tags:
subprocess.run(["git", "push", remote, *tags], check=True)
else:
…How to Revert a Commit and Create a New Revert Commit in Python
Demonstrates a mock Git repository that creates a new revert commit on top of the current head when reverting an existing commit.
class GitCommit:
"""Minimal mock of a git commit for demonstrating revert behavior."""
def __init__(self, sha, message):
self.sha = sha
self.message = message
self.parent = None
class GitRepository:
"""Mock repository tracking a simple commit chain."""
def __init__(self):
…How to sync a fork with upstream in Python
Run git fetch and merge commands from Python with subprocess to sync a forked repository with upstream/main.
import subprocess
import sys
def sync_fork_with_upstream():
"""Simulate syncing a forked repo with upstream via git commands."""
# Mock git operations: pretend to fetch from upstream and merge into main
fetch_result = subprocess.run(
["git", "fetch", "upstream"],
capture_output=True, tex…How to Calculate Cloud Cost Estimates with a Python Dictionary
Mocks a cloud pricing calculator using a dictionary of service rates and computes total estimated cost for given service hours.
def estimate_cost(service, hours, rate_table=None):
if rate_table is None:
rate_table = {
"basic": 50,
"standard": 75,
"premium": 100
}
if service not in rate_table:
raise ValueError(f"Unknown service: {service}")
return rate_table[service] * hour…How to Mock ELB Target Health Status in Python
Simulate AWS Elastic Load Balancer target health checks with a Python dict that mutates status and healthy host counts.
from random import randint
def elb_target_mock_status(target_id, healthy=True):
targets = {
1: {"Id": "i-001", "Status": "healthy", "Port": 80, "HealthyHostCount": 1},
2: {"Id": "i-002", "Status": "unhealthy", "Port": 80, "HealthyHostCount": 0},
3: {"Id": "i-003", "Status": "healthy", "Por…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 Generate a Mock devcontainer.json Config in Python
Build a reproducible devcontainer.json file with Python, composing name, image, extensions, forwarded ports, and a post-create command as a dict.
import json
from pathlib import Path
def create_devcontainer_config(
image: str = "mcr.microsoft.com/devcontainers/python:3.11",
name: str = "python-dev-container",
ports: list[int] | None = None,
post_create: str | None = None,
) -> dict:
config = {
"name": name,
"image": image,
…How to List Pre-commit Hooks from YAML Config in Python
Parse a .pre-commit-config.yaml file with PyYAML and print every hook ID paired with its source repository.
import yaml
pre_commit_config = """
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.5.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- repo: https://github.com/psf/black
rev: 23.11.0
hooks:
- id: black
"""
def list_hooks(c…How to Mock docker compose up Healthcheck in Python
Simulate docker compose up with a healthcheck cycle using Python loops, delays, and simulated service statuses.
import subprocess
import time
def run_healthcheck():
"""Mock a docker compose up with a healthcheck cycle."""
services = ["web", "db", "cache"]
print("Starting docker compose services...")
for service in services:
print(f"[{service}] starting...")
time.sleep(0.1)
print(f"[…pytest mark slow skip integration
Uses pytest markers to select fast tests, skip unfinished ones, and run integration checks with verbose output.
import pytest
def test_fast():
assert 1 + 1 == 2
@pytest.mark.slow
def test_slow():
import time
time.sleep(1)
assert 5 * 5 == 25
@pytest.mark.skip(reason="Not ready for production")
def test_skipped():
assert 2 + 2 == 5
@pytest.mark.integration
def test_integration():
database = {"users": […How to Test HTTPX Async Client Pool Reuse with Mocks in Python
Mock an httpx.AsyncClient to verify connection pool reuse by asserting GET calls share a single client instance across concurrent async requests.
import asyncio
import httpx
from unittest.mock import AsyncMock, patch, Mock
async def fetch_with_pool(client, url, n_reuses=3):
results = []
for i in range(n_reuses):
resp = await client.get(url)
results.append(resp.status_code)
await asyncio.sleep(0) # yield to loop to mimic real us…How to Use pool.map for CPU-Bound Tasks in Python
Distribute CPU-intensive functions across processes with multiprocessing.Pool.map and measure the performance gain.
from multiprocessing import Pool
import time
def cpu_bound_task(n):
"""Mock CPU-bound work: compute sum of squares."""
total = 0
for i in range(n):
total += i * i
return total
if __name__ == "__main__":
numbers = [10_000_000, 12_000_000, 8_000_000, 15_000_000]
start = time.perf_count…How to Use threading.Lock to Synchronize a Counter in Python
Safely increment a shared counter across multiple threads using threading.Lock as a mutex to prevent race conditions.
import threading
counter = 0
lock = threading.Lock()
def increment():
global counter
for _ in range(100000):
with lock:
counter += 1
threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
print(f"Final counter valu…Generate Fake User Data with Faker in Python
Use the Faker library to generate realistic fake user profiles with names, emails, phone numbers, and addresses for tests or demos.
from faker import Faker
fake = Faker()
def generate_user():
return {
"name": fake.name(),
"email": fake.email(),
"phone": fake.phone_number(),
"address": fake.address().replace("\n", ", "),
}
if __name__ == "__main__":
user = generate_user()
for key, value in user.ite…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.