Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
How to Pipe Data Through a List of Transform Functions in Python
Applies a sequence of functions to an initial value using functools.reduce, creating a reusable pipe utility.
from functools import reduce
def pipe(data, *transforms):
return reduce(lambda value, func: func(value), transforms, data)
def double(x):
return x * 2
def add_one(x):
return x + 1
def to_string(x):
return f"Result: {x}"
if __name__ == "__main__":
initial = 5
result = pipe(initial, double, …Build a Simple ETL Pipeline in Python
A simple ETL pipeline that reads JSON Lines, transforms records with filtering and normalization, and writes the result to JSON.
import json
from pathlib import Path
def read_input(file_path: Path) -> list[dict]:
"""Read JSON lines file into list of dicts."""
with file_path.open("r", encoding="utf-8") as f:
return [json.loads(line) for line in f if line.strip()]
def transform(records: list[dict]) -> list[dict]:
"""Transf…How to Chunk a Long Document for RAG Retrieval in Python
Split text into overlapping chunks at sentence boundaries using a custom Python function suitable for RAG retrieval pipelines.
import re
from pathlib import Path
def chunk_document(text, chunk_size=500, overlap=100):
"""Split text into overlapping chunks suitable for RAG retrieval."""
# Normalize whitespace
text = re.sub(r'\s+', ' ', text).strip()
chunks = []
start = 0
while start < len(text):
end = min(s…How to build a mock RAG pipeline in Python
Build a minimal Retrieval-Augmented Generation pipeline that retrieves the best-matching document by keyword overlap and generates a template-based answer.
def simple_rag_pipeline(question, documents):
"""
A minimal mock RAG pipeline: retrieve relevant context, then generate an answer.
"""
# Step 1: Retrieve — mock retrieval by simple keyword scoring
scores = []
for doc in documents:
doc_words = set(doc.lower().split())
question_wo…Create Mock Watermarked Image Bytes in Python Without PIL
Builds a mock image-like byte stream with an embedded watermark using only stdlib modules, for testing pipelines without PIL.
from io import BytesIO
import zlib
import struct
def create_watermarked_bytes(width: int, height: int, watermark: bytes) -> bytes:
"""Create a mock image-like byte stream with a watermark (no PIL)."""
header = struct.pack("<2I", width, height)
payload = watermark * max(1, (width * height // max(1, len(wa…Count Records Processed per Category in Python
Use a Counter dictionary to track how many records of each type (ok, error, retry) were processed in a data pipeline.
from collections import Counter
import random
processed_counter = Counter()
def process_records(records):
for record in records:
processed_counter[record] += 1
return len(records)
if __name__ == "__main__":
sample_records = [random.choice(["ok", "error", "retry"]) for _ in range(10)]
print(f…Create Data Helper Functions in Python for Beginners
Build reusable Python helper functions to load, filter, sort, summarize, and save JSON data — a beginner-friendly starting point for small data pipelines.
import json
from pathlib import Path
from typing import Any, Dict, List
def load_json_file(filepath: str) -> Dict[str, Any]:
"""Load JSON data from a file."""
with Path(filepath).open("r", encoding="utf-8") as file:
return json.load(file)
def filter_by_key(
data: List[Dict[str, Any]], key: str,…ETL in Python: Extract CSV, Transform Dict, Load JSON
Build a simple ETL pipeline in Python that reads a CSV file, transforms each row (stripping whitespace and converting numeric fields), and writes the result to JSON.
import csv
import json
from pathlib import Path
def extract_csv(file_path):
"""Read CSV file and return list of row dictionaries."""
with Path(file_path).open('r', newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
return list(reader)
def transform_dicts(rows):
"""Transform ro…ETL in Python: Extract CSV, Transform Dicts, Load JSON
Build a simple ETL pipeline that reads a CSV, normalizes keys and converts price to float, then writes structured JSON.
import csv
import json
from pathlib import Path
def etl_csv_to_json(csv_path: str, json_path: str) -> None:
"""Extract CSV, transform rows to dicts, load to JSON."""
with open(csv_path, mode='r', newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
records = list(reader)
# Trans…Enrich Events with Geo IP Data in Python
Returns a copy of each event dictionary, enriched with a geo-location dict from a mock IP-to-geo lookup table, with a fallback for unknown IPs.
import ipaddress
GEO_IP_DB = {
"192.168.1.10": {"country": "US", "city": "New York", "lat": 40.7128, "lon": -74.0060},
"10.0.0.5": {"country": "DE", "city": "Berlin", "lat": 52.5200, "lon": 13.4050},
"172.16.0.8": {"country": "JP", "city": "Tokyo", "lat": 35.6762, "lon": 139.6503},
}
EVENTS = [
{"id…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_…Filter Records by Required Fields in Python
Filter a list of dictionaries, keeping only records where every required field is present and not None.
def filter_records(records, required_fields):
"""Return only records that have all required fields non-null."""
return [
record for record in records
if all(record.get(field) is not None for field in required_fields)
]
if __name__ == "__main__":
sample_records = [
{"name": "Al…Generate a Deterministic Hash for Deduplication in Python
Create a stable SHA-256 fingerprint from nested data and file contents to deduplicate records in a data pipeline.
import hashlib
import json
from pathlib import Path
def natural_key_hash(data, salt=""):
"""
Generate a deterministic fingerprint from raw data (dict/list/str).
Uses JSON canonical-ish serialization with sorted keys and SHA-256.
"""
canonical = json.dumps(data, sort_keys=True, separators=(",", ":"…Generate a Mock CDC Changelog in Python
Simulate a CDC changelog with INSERT, UPDATE, and DELETE operations, timestamps, and record snapshots for testing data pipelines.
import json
from datetime import datetime, timedelta
def generate_mock_changelog(records, operations=("INSERT", "UPDATE", "DELETE")):
"""Simulate a CDC changelog from a list of record snapshots."""
base_time = datetime(2025, 1, 1, 8, 0, 0)
changelog = []
for idx, record in enumerate(records):
…How to Build Data Processing Functions in Python
Create reusable helper functions to load, filter, transform, and aggregate CSV data in Python.
import csv
from pathlib import Path
def load_data(filepath):
"""Load CSV data into a list of dicts."""
with open(filepath, "r", newline="", encoding="utf-8") as f:
return list(csv.DictReader(f))
def filter_rows(rows, column, value):
"""Keep rows where column equals value."""
return [row for…How to Build a Simple Data Pipeline in Python
A beginner-friendly data pipeline that loads JSON, filters records by a field value, and aggregates counts per category.
import json
from pathlib import Path
def load_json(filepath: str | Path) -> list[dict]:
"""Load a JSON file containing a list of records."""
with Path(filepath).open("r", encoding="utf-8") as f:
return json.load(f)
def filter_records(records: list[dict], field: str, value) -> list[dict]:
"""Kee…How to Clean and Format Data in Python
This code loads JSON data, cleans records by removing empty fields and normalizing text, then summarizes the results with counts and unique keys.
import json
from pathlib import Path
def load_data(filepath: str) -> dict:
"""Load JSON data from a file."""
with Path(filepath).open("r", encoding="utf-8") as f:
return json.load(f)
def clean_records(records: list[dict]) -> list[dict]:
"""Remove empty fields and normalize text to lowercase."""…How to Compress Pipeline Output Gzip Per Partition in Python
Compress each partition of pipeline output into a separate gzip file and verify the compressed data by reading it back.
import gzip
import io
import random
from pathlib import Path
def compress_partition(partition_data: list[str], output_path: Path) -> int:
"""Compress a partition of data to a gzip file, returns bytes written."""
with gzip.open(output_path, 'wt', encoding='utf-8') as f:
f.writelines(partition_data)
…How to Convert Data Types in a Python Data Pipeline
Demonstrates a simple Python data pipeline that converts string values to proper types (bool, int, float, datetime) and outputs structured JSON.
import json
from datetime import datetime
def convert_value(value):
"""Convert string values to appropriate Python types."""
if value.lower() == "true":
return True
if value.lower() == "false":
return False
if value.isdigit():
return int(value)
try:
return float(val…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 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…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.