Python Code
Samples
Easy snippets you can copy, study, and run in the browser editor.
How to Import Users from CSV into LDAP-like Dicts in Python
Reads a CSV of user records and converts each row into an LDAP-style dictionary with standard attributes using Python's csv module.
import csv
import io
from pathlib import Path
def mock_ldap_import(csv_path):
"""
Reads a CSV file with user data and returns a list of LDAP-like user dicts.
Adds standard LDAP attributes that would come from directory schema.
"""
with open(csv_path, newline="", encoding="utf-8") as csvfile:
…How to Perform a DNS Lookup for A Records in Python
Resolve a hostname to IPv4 A records using Python's built-in socket.getaddrinfo and return a sorted list of addresses.
import socket
def get_a_records(hostname):
"""Fetch A records (IPv4 addresses) for a given hostname."""
try:
# getaddrinfo with family AF_INET restricts to IPv4 (A records)
infos = socket.getaddrinfo(hostname, None, socket.AF_INET)
# Each info tuple: (family, type, proto, canonname, so…How to Sort Command-Line Arguments in Python
Build a beginner-friendly argparse CLI that sorts numbers or words passed as arguments, with an optional reverse flag.
import argparse
def main():
parser = argparse.ArgumentParser(description="Sort numbers or words from the command line.")
parser.add_argument("items", nargs="+", help="Items to sort (numbers or words)")
parser.add_argument("--reverse", "-r", action="store_true", help="Sort in descending order")
args =…Attach Source File Metadata to Records in Python
Add a source filename field to each record in a list by merging a new key into every dictionary using a dict unpacking comprehension.
from pathlib import Path
import json
def attach_source_metadata(records, source_file):
"""Attach source filename metadata to each record."""
return [
{**record, "source": Path(source_file).name}
for record in records
]
if __name__ == "__main__":
source = "/data/raw/customers.csv"
…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…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=(",", ":"…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 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 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 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 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…How to Design a Cloud Data Helper Class in Python
A beginner-friendly Python helper class that saves, loads, and aggregates JSON records locally, simulating cloud-style data handling.
import json
from pathlib import Path
from datetime import datetime
class CloudDataHelper:
"""Beginner-friendly helper for working with cloud-based JSON data."""
def __init__(self, base_dir="cloud_data"):
self.base_dir = Path(base_dir)
self.base_dir.mkdir(exist_ok=True)
def save_record(s…How to Capture Logging Records with pytest caplog in Python
Capture and assert on logging records in pytest using the built-in caplog fixture.
import logging
import pytest
def divide(a, b):
"""Divide two numbers and log an error if b is zero."""
if b == 0:
logging.error("Division by zero attempted")
return None
logging.info(f"Dividing {a} by {b}")
return a / b
def test_divide_logs_error(caplog):
with caplog.at_level(logg…NamedTuple typed record in Python
Define a lightweight immutable record with type hints using typing.NamedTuple; access fields by name and unpack like a tuple.
from typing import NamedTuple
class Point(NamedTuple):
x: float
y: float
label: str = "origin"
if __name__ == "__main__":
p = Point(3.5, -2.0, "A")
print(p)
print(f"x={p.x}, y={p.y}, label={p.label}")
print("is tuple:", isinstance(p, tuple))
q = Point(1.0, 1.0)
print(q)
# …How to Implement the Repository Pattern in Python with an In-Memory Dict
Stores, retrieves, updates, and deletes user records in memory using a Repository abstraction over a plain dict, isolating data access from business logic.
class UserRepository:
def __init__(self):
self._storage = {}
self._next_id = 1
def create(self, name, email):
user_id = self._next_id
self._next_id += 1
self._storage[user_id] = {"id": user_id, "name": name, "email": email}
return self._storage[user_id]
def…How to Build a Data Helper Class in Python for Beginners
Create a beginner-friendly DataHelper class that stores, retrieves, filters, and summarizes records in a list of dictionaries.
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
@dataclass
class DataHelper:
"""A beginner-friendly helper for common data tasks."""
data: List[Dict[str, Any]] = field(default_factory=list)
def add_record(self, record…How to Build a Simple Filter Helper in Python for API Design
Create a reusable data filter service with dataclasses that mimics gRPC request/response patterns for filtering dataset records.
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
@dataclass
class FilterRequest:
"""A simple filter request mirroring a gRPC message structure."""
field_name: str
operator: str # eq, ne, gt, lt, contains
value: Any
page_size: int = 10
page_token: Optional…At Most Once Fire-and-Forget Mock in Python
A Python mock that enforces send() is called at most once and records the arguments for verification.
class FireForgetMock:
def __init__(self):
self._calls = 0
self._last_args = None
self._last_kwargs = None
def send(self, *args, **kwargs):
if self._calls > 0:
raise RuntimeError("send() called more than once")
self._calls += 1
self._last_args = args
…How to Build a Message Stream Queue in Python
A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.
from collections import deque
import time
class MessageStream:
def __init__(self, messages):
self._queue = deque(messages)
self._sent = []
def send_next(self):
if not self._queue:
return None
message = self._queue.popleft()
self._sent.append(message)
…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.