Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
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 Track Checkpoint Offset After Batch Commit in Python
A batch processor that tracks the last successfully committed offset after processing records in batches, advancing the checkpoint only when each batch commits successfully.
import json
from typing import Any
class BatchProcessor:
"""Tracks checkpoint offset after committing batches."""
def __init__(self, batch_size: int = 3):
self.batch_size = batch_size
self.offset = 0 # last successfully committed offset (exclusive)
self.total_committed = 0
def …How to Unpivot Wide to Long with pandas melt in Python
This code demonstrates how to use pandas.melt to unpivot a wide DataFrame into a tidy long format, converting subject columns into rows.
import pandas as pd
# Sample wide-format data
df_wide = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
'math': [90, 85, 95],
'science': [80, 92, 88]
})
print("Original wide DataFrame:")
print(df_wide)
# Melt: unpivot subject columns into rows
df_long = pd.melt(
df_wide,
…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 detect anomalies in a column using z-score in Python
Detect outliers in a list of numbers using z-score statistics, flagging values that deviate significantly from the mean.
import random
def z_score_anomaly_detection(data, threshold=2.0):
"""
Detect anomalies in a list of numbers using z-score.
"""
mean = sum(data) / len(data)
variance = sum((x - mean) ** 2 for x in data) / len(data)
std_dev = variance ** 0.5
if std_dev == 0:
return []
a…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…How to shard output by primary key hash mod N in Python
This code computes a consistent shard index for any primary key string using an MD5 hash mod the number of shards, enabling stable key-based data distribution.
import hashlib
def shard_id(primary_key: str, num_shards: int) -> int:
"""Return the shard index for a primary key using MD5 hash mod N."""
digest = hashlib.md5(primary_key.encode("utf-8")).hexdigest()
hash_int = int(digest, 16)
return hash_int % num_shards
if __name__ == "__main__":
keys = ["use…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)
…Implement Exactly-Once Transaction Log in Python
A mock transaction log that deduplicates transaction IDs so each is recorded only once, with a dataclass for records and simple in-memory storage.
from dataclasses import dataclass
from typing import Dict, Optional
@dataclass
class TxnRecord:
txn_id: str
status: str
class ExactlyOnceTxnLog:
def __init__(self) -> None:
self._log: Dict[str, TxnRecord] = {}
self._processed_ids: set = set()
def record(self, txn_id: str, status: s…Parallel Extract Multiple Sources with Threads in Python
Extract data from multiple sources in parallel using ThreadPoolExecutor and verify results match sequential processing.
import threading
from concurrent.futures import ThreadPoolExecutor
def extract_from_source(source):
"""Simulate extracting data from a source."""
return f"Data from {source}"
def main():
sources = ["source_a", "source_b", "source_c", "source_d"]
# Sequential extraction for comparison
sequent…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 =…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(…Union Multiple DataFrames with Aligned Columns in Python
Concatenate DataFrames with different columns, aligning them and filling missing values with NaN using pandas concat.
import pandas as pd
from io import StringIO
# Sample dataframes with different columns
df1 = pd.DataFrame({
'id': [1, 2, 3],
'name': ['Alice', 'Bob', 'Charlie'],
'age': [25, 30, 35]
})
df2 = pd.DataFrame({
'id': [4, 5],
'name': ['Diana', 'Eve'],
'city': ['NYC', 'LA']
})
df3 = pd.DataFrame({
…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")
…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.…Bisect Good Bad Automation Script in Python
This Python script implements a binary search to find the first bad version in a list, simulating an automation script for git bisect.
import bisect
def find_first_bad(versions):
"""Given a list of version objects with .is_bad(), find first bad version."""
lo, hi = 0, len(versions)
while lo < hi:
mid = (lo + hi) // 2
if versions[mid].is_bad():
hi = mid
else:
lo = mid + 1
return lo
clas…Build a Simple Log Graph in Python
Create a basic one-dimensional bar chart from log lines by counting occurrences of leading numeric keys.
import heapq
def log_graph(log_lines: list[str]) -> str:
"""Build a simple per-line, one-dimensional visual graph from log entries."""
counts: dict[int, int] = {}
for line in log_lines:
tokens = line.split()
if tokens:
try:
idx = int(tokens[0])
exce…Bump Semantic Version Git Tag in Python
Automatically find the latest Git tag and compute the next patch release using semantic versioning (semver) in Python.
from re import match
from subprocess import run
SEMVER_PATTERN = r"^v(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)(?:-(?P<prerelease>[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?(?:\+(?P<buildmetadata>[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?$"
def get_latest_tag() -> str:
result = run(["git", "describe…Count Unique Contributors from Git Shortlog in Python
Parses git shortlog -sn output to count the number of unique contributors, handling duplicate entries and variable whitespace.
import subprocess
from collections import Counter
# Mock shortlog output as a list of lines (simulating git shortlog -sn output)
MOCK_SHORTLOG = """ 120 Alice Johnson
88 Bob Smith
45 Alice Johnson
30 Carol Williams
25 Bob Smith
10 Dave Brown
"""
def count_contributors_from_shortlog(text):
"…Create a Mock GitHub Release API in Python for Testing gh CLI
Build an in-memory GitHub Releases API mock that mimics create_release and list_releases for unit testing gh CLI stubs without network calls.
import json
from unittest.mock import patch, Mock
class GitHubReleaseAPI:
"""Mock GitHub Releases API for testing gh CLI stub behavior."""
def __init__(self):
self.releases = {}
self.counter = 1
def create_release(self, repo, tag, name=None, notes=None):
release_id = self…Detect Merge Conflict Markers in a File with Python
Scan a file line by line to detect Git merge conflict markers (<<<<<<<, =======, >>>>>>>) and report their line numbers with context.
from pathlib import Path
def detect_merge_conflicts(file_path):
conflicts = []
with open(file_path, 'r') as f:
lines = f.readlines()
for i, line in enumerate(lines, 1):
if line.startswith('<<<<<<<'):
conflict_marker = 'conflict start'
conflicts.append((i, confl…Fetch Pull Rebase Workflow Script in Python
A Python script that automates the git fetch, checkout, and pull with rebase workflow using subprocess.
import subprocess
import sys
def run_git_command(args: list[str]) -> str:
"""Run a git command and return its stdout, or raise on failure."""
result = subprocess.run(
["git", *args],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
prin…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.