Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
Restore sqlite from latest backup file in Python
This script finds the most recently modified backup file in a directory and restores it to the main database path, then verifies the restored data.
import sqlite3
import glob
import os
import shutil
def restore_latest_backup(db_path, backup_dir):
backups = sorted(glob.glob(os.path.join(backup_dir, "*.db")), key=os.path.getmtime)
if not backups:
raise FileNotFoundError("No backup files found")
latest = backups[-1]
shutil.copy2(latest, db_p…Restrict Secrets File Permissions with the chmod Script in Python
This script restricts a secrets file to 0600 permissions, rotates it to a dated backup, and creates a fresh protected file for secure automation workflows.
import os
import sys
import stat
from pathlib import Path
def restrict_secrets_file(filepath: str) -> None:
"""Set restrictive permissions (0600) on a secrets file."""
path = Path(filepath).expanduser()
if not path.is_file():
raise FileNotFoundError(f"Secrets file not found: {path}")
…Stress CPU Threads with a Mock Compute in Python
Simulates CPU-intensive work across multiple threads to test how Python schedules parallel compute.
import threading
import time
def stress_cpu(iterations: int):
result = 0
for i in range(iterations):
result += i * i % 1000
return result
def run_mock_stress(thread_count: int, iterations: int):
threads = []
for tid in range(thread_count):
t = threading.Thread(target=lambda: str…Toggle VPN Mock Network Manager Script in Python
Simulate a VPN manager with connect, disconnect, toggle, and status methods for testing or demo workflows.
import time
class MockVPNManager:
def __init__(self):
self.is_connected = False
self.servers = ["us-west", "eu-central", "asia-east"]
self.active_server = None
def toggle(self):
if self.is_connected:
self.disconnect()
else:
self.connect()
d…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 Deduplicate Events with At-Least-Once Delivery in Python
Implements an exactly-once processing pattern for at-least-once event delivery by tracking seen event IDs in a set, skipping duplicates.
seen_ids = set()
def process_event(event_id: str, payload: dict) -> dict:
"""Process an event exactly once, ignoring duplicates."""
if event_id in seen_ids:
return {"status": "duplicate", "event_id": event_id}
seen_ids.add(event_id)
return {"status": "processed", "event_id": event_id, **payloa…How to Explode an Array Field into Multiple Rows in Python
This code flattens a list of dictionaries by exploding each array field value into its own row, duplicating the other fields as needed.
from collections import defaultdict
data = [
{"id": 1, "name": "Alice", "tags": ["python", "data", "ai"]},
{"id": 2, "name": "Bob", "tags": ["web", "devops"]},
{"id": 3, "name": "Carol", "tags": []},
]
def explode_array_field(records, array_field):
result = []
for record in records:
for v…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…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.