Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Count JSON Records in Python
Read a JSON file and count the number of top-level records, handling both list and dictionary structures.
import json
from pathlib import Path
def count_records(json_file):
"""Count top-level records in a JSON file."""
with open(json_file, "r") as f:
data = json.load(f)
# Handle both list of records and dict of records
if isinstance(data, list):
return len(data)
elif isinstance(da…How to Group Data by Key in Python
Group a list of dictionaries by a specified key using a defaultdict and compute per-group averages.
from collections import defaultdict
def group_by_key(data, key):
grouped = defaultdict(list)
for item in data:
grouped[item[key]].append(item)
return dict(grouped)
if __name__ == "__main__":
records = [
{"name": "Alice", "dept": "Engineering", "score": 85},
{"name": "Bob", "de…How to Hash Email Addresses in a PII Masking Pipeline in Python
Replaces every email address in a text string with its SHA-256 hash to protect personally identifiable information (PII).
import hashlib
import re
def hash_email(email: str) -> str:
"""Mask an email address by hashing it with SHA-256."""
normalized = email.strip().lower()
return hashlib.sha256(normalized.encode("utf-8")).hexdigest()
def mask_pii_emails(text: str) -> str:
"""Replace all email addresses in text with their…How to Implement Slowly Changing Dimension Type 2 History in Python
Build a type-2 slowly changing dimension pipeline that closes old records and opens new ones when customer data changes.
from datetime import datetime, timedelta
def apply_scd_type2(records, current_date):
"""Returns active records after inserting new records with type-2 history."""
history = []
active = {}
for record in records:
key = record["customer_id"]
if key in active:
active[key]["end…How to List Failed Records in a Dead Letter Queue Mock in Python
A mock Dead Letter Queue stores failed processing records with error details and timestamps, lists them, and exports to JSON.
import json
from datetime import datetime, timedelta
import random
class DeadLetterQueue:
def __init__(self):
self.failed_records = []
def add_failed_record(self, record_id, payload, error_message):
self.failed_records.append({
"record_id": record_id,
"payload": paylo…How to Merge Incremental Snapshot Upsert Dict in Python
Merge a snapshot dict into a base dict, recursively updating nested dictionaries while preferring snapshot values on conflicts.
def merge_upsert(base: dict, snapshot: dict) -> dict:
"""
Merge a snapshot dict into a base dict, preferring snapshot values
on key conflicts (upsert semantics). Nested dicts are merged recursively.
"""
result = dict(base)
for key, value in snapshot.items():
if key in result and i…How to Merge Multiple Data Sources in Python
A beginner-friendly helper that merges lists of dictionaries from multiple sources into one combined list using key filtering.
import json
def merge_pipeline_data(*data_sources, keys=()):
"""Merge multiple data sources (list of dicts) into a single list of merged dicts.
Args:
*data_sources: One or more lists of dictionaries.
keys: Tuple of keys to include from each source (empty means all keys).
Returns:
…How to Process CSV Data in Python with a Data Helper
Build a beginner-friendly data helper in Python that loads a CSV file, filters rows by a condition, and summarizes numeric fields.
import csv
from pathlib import Path
DATA = [
{"name": "Alice", "score": 88, "passed": True},
{"name": "Bob", "score": 42, "passed": False},
{"name": "Carol", "score": 95, "passed": True},
]
def load_csv(file_path: Path) -> list[dict]:
with file_path.open(newline="", encoding="utf-8") as f:
r…How to Run a Mock Cron Pipeline Scheduler in Python
This code schedules a mock pipeline job to run every 2 seconds and hourly at :30 using the schedule library, then runs pending tasks for 10 seconds.
import time
import schedule
from datetime import datetime
def run_pipeline():
print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Pipeline executed")
schedule.every(2).seconds.do(run_pipeline)
schedule.every().hour.at(":30").do(run_pipeline)
print("Scheduler started. Press Ctrl+C to stop.")
end_time = ti…How to Sort a List of Dictionaries by Key in Python
A reusable helper function that sorts a list of dictionaries by a specified key, with optional descending order support.
from typing import List
def sort_records(records: List[dict], key: str, descending: bool = False) -> List[dict]:
"""Sort a list of dictionaries by a specified key."""
return sorted(records, key=lambda record: record[key], reverse=descending)
def demonstrate_sorting() -> None:
users = [
{"name": …How to Stream a Large JSONL File Line by Line in Python
Process a large JSON-lines file incrementally using streaming techniques to avoid loading the entire file into memory.
import json
def process_large_file(filepath, chunk_size=8192):
"""
Stream a large JSON-lines file line by line, processing each record
without loading the entire file into memory.
"""
total_count = 0
total_sum = 0
with open(filepath, 'r') as f:
while True:
chunk = …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)…How to create a dated snapshot path for a dataset in Python
Generate a versioned directory path combining a base directory, dataset name, and today's date, ready for creating snapshots in data pipelines.
import datetime
import os
from pathlib import Path
def snapshot_path(base_dir: str, dataset_name: str) -> Path:
"""Return a dated snapshot path for a dataset under a base directory."""
today = datetime.date.today().isoformat()
return Path(base_dir) / dataset_name / today
if __name__ == "__main__":
…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…Idempotent Pipeline Dedupe by Record ID Set in Python
Filters records against a persistent set of seen IDs, returning only new ones and the updated set for idempotent pipeline processing.
def dedupe_records(records, seen_ids=None):
"""Return records whose id has not been seen before."""
if seen_ids is None:
seen_ids = set()
unique = []
for record in records:
record_id = record.get("id")
if record_id not in seen_ids:
seen_ids.add(record_id)
…Map Partition Over Chunks in Python with Multiprocessing and Mock
Process data in chunks across multiple CPU cores using multiprocessing Pool.map, and mock the chunk function to test partitioning behavior without heavy computation.
from multiprocessing import Pool
from unittest.mock import patch, Mock
def process_chunk(chunk):
return [x * x for x in chunk]
def map_partition_over_chunks(data, chunk_size, process_func=process_chunk):
chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
with Pool() as pool:
…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
…Test a Python Pipeline with Fixture Sample Rows
Test pipeline functions with sample rows provided by a pytest fixture, verifying required keys and value constraints.
import pytest
def get_value(data: dict, key: str):
return data.get(key)
def sample_rows():
return [
{"name": "Alice", "age": 30, "city": "London"},
{"name": "Bob", "age": 25, "city": "Paris"},
{"name": "Charlie", "age": 35, "city": "Berlin"},
]
@pytest.fixture
def sample_data(…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…Validate dict schema at pipeline boundary in Python
This code validates a dictionary against a TypedDict schema at a pipeline boundary, enforcing required fields and types with custom error messages.
from typing import Any, TypedDict
class Person(TypedDict):
name: str
age: int
email: str
def validate_person(data: dict[str, Any]) -> Person:
errors: list[str] = []
if not isinstance(data.get("name"), str) or not data["name"].strip():
errors.append("name must be a non-empty string")
…How to Mock a Fast uv pip sync in Python
Simulate a fast uv pip sync by mocking file operations and subprocess calls to test dependency installation workflows.
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
def uv_pip_sync_fast_install_mock(requirements_text: str) -> dict:
"""Simulate a fast uv pip sync by mocking file operations and subprocess calls."""
mock_dir = Path(tempfile.mkdtemp(prefix="uv_mock_"))
req_lines…How to Mock a pipx Install Command in Python
Simulate a pipx install step by validating tool names and printing the exact command output a real pipx run would produce.
import subprocess
import sys
def install_with_pipx(tool_name: str) -> str:
"""
Mock a pipx install step by validating the tool name and
simulating the installation command output.
"""
allowed_tools = {"black", "flake8", "mypy", "ruff"}
if tool_name not in allowed_tools:
raise ValueErr…Mock pip-compile to Resolve Requirements in Python
A mock function that mimics pip-compile by converting a requirements.in file into pinned, locked package versions.
import subprocess
import tempfile
from pathlib import Path
def compile_requirements_mock(requirements_in: str) -> str:
"""Mock pip-compile: resolve a simple requirements.in into a locked format."""
lines = [line.strip() for line in requirements_in.splitlines() if line.strip() and not line.startswith("#")]
…How to Send and Receive Messages Between Processes with multiprocessing.Pipe in Python
Use multiprocessing.Pipe to create a two-way connection between two processes, send a message from parent to child, and receive a reply back.
import multiprocessing
def child_process(conn):
"""Receive from parent and send back a response."""
message = conn.recv()
print(f"Child received: {message}")
conn.send("Hello from child!")
if __name__ == "__main__":
parent_conn, child_conn = multiprocessing.Pipe()
process = multiprocessing…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.