Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
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 Incremental Load with Watermark by updated_at in Python
Load only new or changed rows into SQLite by comparing an updated_at timestamp against a stored watermark, returning counts and the new watermark.
import sqlite3
from datetime import datetime, timedelta
def watermark_incremental_load(db_path, table_name, last_watermark, source_data):
"""Load only rows with updated_at greater than the last watermark."""
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Create table if it doesn't exist
…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 Parse Data in Python: A Beginner's Helper
This helper parses a JSON payload, extracts user names, emails, and signup dates, then summarizes the results.
import json
from datetime import datetime
from typing import Dict, List
def parse_data(payload: str) -> Dict[str, List]:
"""Parse a JSON payload and extract useful fields."""
raw = json.loads(payload)
users = raw.get("users", [])
parsed = {
"names": [],
"emails": [],
"signup_…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 Register a Dataset Schema as JSON in Python
Define a catalog of dataset schemas and serialize them to JSON with the standard library json module.
import json
catalog = {
"name": "sample_catalog",
"version": "1.0",
"datasets": [
{
"id": "users",
"type": "table",
"fields": [
{"name": "id", "type": "integer", "key": True},
{"name": "email", "type": "string", "nullable": False}…How to Safely Coerce Strings to Numbers in Python
A safe conversion function that turns strings into integers or floats, returning a fallback value when conversion fails.
import math
def to_number(value, fallback=None):
"""Safely coerce a string to int or float, returning fallback on failure."""
if isinstance(value, (int, float)):
return value
try:
# Try int first for clean whole numbers
return int(value)
except (ValueError, TypeError):
…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 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…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({
…How to Make a Shallow Clone of an Object in Python
Demonstrates using copy.copy() to create a shallow clone of a Python object, showing how nested mutable data is shared while top-level attributes are independent.
import copy
class Config:
def __init__(self):
self.settings = {"volume": 50}
self.user = "admin"
def demonstrate_shallow_copy():
original = Config()
shallow = copy.copy(original)
# Mutating nested object is visible in both (shallow copy share it)
shallow.settings["volume"] = 90…Upload Assets to GitHub Release with Python Mock
Simulates uploading binary and text assets to a GitHub release using a mock server, returning structured metadata for each upload.
import json
import os
import tempfile
from datetime import datetime
class ReleaseUploader:
"""Simulates uploading assets to a release with a mock server."""
def __init__(self, owner: str, repo: str, tag: str):
self.owner = owner
self.repo = repo
self.tag = tag
self.uploade…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.