Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Serialize Chat Messages to a JSON File in Python
Writes a list of chat message dicts to a JSON file with metadata like export time and message count.
import json
from pathlib import Path
from datetime import datetime
def serialize_messages(messages, output_path):
data = {
"exported_at": datetime.now().isoformat(),
"count": len(messages),
"messages": messages
}
Path(output_path).write_text(
json.dumps(data, indent=2, ensu…How to Validate JSON Output Against a Dict Schema in Python
Validate JSON-like data against a simple dict schema with type checking and descriptive error messages using only the Python standard library.
from typing import Dict, Any, List, Union
def validate_json(data: Any, schema: Dict[str, str]) -> List[str]:
"""
Validate JSON-like data against a simple dict schema.
Schema format: {field_name: expected_type} where type is one of:
'str', 'int', 'float', 'bool', 'list', 'dict', 'any'
Returns list …How to randomly assign a prompt variant to each key in Python
Randomly pick one variant from a list for each prompt key, useful for A/B testing message variations.
import random
def assign_prompt_variant(prompts: dict[str, list[str]]) -> dict[str, str]:
"""Assign a random prompt variant to each prompt key."""
return {key: random.choice(variants) for key, variants in prompts.items()}
if __name__ == "__main__":
prompt_bank = {
"greeting": ["Hello!", "Hi there…How to Detect Network Interface Changes in Python
Monitor active network interfaces and print a message when an interface is added or removed using psutil and socket.
import socket
import psutil
import time
def get_network_interfaces():
"""Return a set of currently active interface names."""
active_ifaces = set()
for iface, addrs in psutil.net_if_addrs().items():
for addr in addrs:
if addr.family == socket.AF_INET: # IPv4 address present
…Monitor Disk Usage and Alert in Python
A Python script that checks disk usage percentage against a threshold and returns an ALERT or OK message with free space details.
import shutil
import os
def check_disk_usage(path="/", threshold=85.0):
usage = shutil.disk_usage(path)
percent_used = (usage.used / usage.total) * 100
if percent_used > threshold:
return (f"ALERT: Disk usage at {percent_used:.1f}% on {path} "
f"(exceeds {threshold}% threshold…Post a message to a Slack webhook in Python
Send a message to a Slack webhook endpoint using the standard library's urllib.request, handling the POST request and response cleanly.
import json
from urllib import request
def post_to_slack(webhook_url: str, message: str) -> dict:
payload = json.dumps({"text": message}).encode("utf-8")
req = request.Request(
webhook_url,
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
wit…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.…How to Auto-Suggest a SemVer Bump From Git Commit Messages in Python
This code scans Git commit messages (recent or sample) and suggests the next Semantic Versioning bump type — major, minor, patch, or none.
import re
import subprocess
from pathlib import Path
def get_commit_messages(path="."):
"""Read commit messages from a repo or use sample messages."""
if (Path(path) / ".git").exists():
out = subprocess.run(
["git", "-C", path, "log", "--pretty=%s"], capture_output=True, text=True
…How to Create a Git Commit with Message Template in Python
Run a git commit from Python using a standardized message template built from a commit type and description.
import subprocess
import sys
def commit_with_template(commit_type: str, description: str) -> None:
message = f"{commit_type}: {description}"
try:
subprocess.run(["git", "commit", "-m", message], check=True)
print(f"Committed: {message}")
except subprocess.CalledProcessError as e:
…How to Generate Release Notes from Git Commit Messages in Python
This script fetches recent Git commit messages using conventional commit prefixes (feat, fix, etc.), categorizes them, and prints formatted release notes with today's date.
import subprocess
import re
from datetime import datetime
def get_git_log(since_tag="HEAD~10", format_str="%s"):
"""Retrieve commit messages from git log."""
try:
result = subprocess.run(
["git", "log", f"--since={since_tag}", f"--format={format_str}"],
capture_output=True,
…How to Mock AWS SQS Send Receive Delete in Python
Build an in-memory mock of the SQS send, receive, and delete message flow for local testing.
import json
from collections import deque
from uuid import uuid4
class MockSQSQueue:
def __init__(self, name):
self.name = name
self._messages = deque()
self._in_flight = {}
def send_message(self, body, attributes=None):
message_id = str(uuid4())
message = {
…How to Validate Data Fields and Types in Python
Validate required fields and type correctness in a Python dictionary with small helper functions, returning a list of clear error messages.
import json
from typing import Any, Dict, List
def validate_data(data: Dict[str, Any], required_fields: List[str]) -> List[str]:
"""Check required fields exist and are non-empty. Return list of errors."""
errors = []
for field in required_fields:
value = data.get(field)
if value is None o…Mock Google Pub/Sub publish and pull in Python
A lightweight in-memory mock of Google Pub/Sub with publisher/subscriber classes to test topic-based fan-out and message pulling without real infrastructure.
import json
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Any, Callable
@dataclass
class Message:
data: str
attributes: dict[str, str] = field(default_factory=dict)
message_id: str | None = None
ack_id: str | None = None
class MockPublisher:
…Mock SNS publish subscribe fanout in Python
Simulates AWS SNS publish/subscribe with an in-memory topic-to-endpoints dict that fans out messages to all subscribers.
class SNSMock:
def __init__(self):
self.topics = {}
def create_topic(self, name):
if name not in self.topics:
self.topics[name] = []
return f"arn:aws:sns:us-east-1:123456789012:{name}"
def subscribe(self, topic_name, endpoint):
self.topics.setdefault(topic_name…How to Create a Rich Console Progress Bar Mock in Python
This code uses Rich's Console and Progress API to build a simulated progress bar for a long-running task, updating progress and printing status messages.
import time
from rich.console import Console
from rich.progress import Progress, BarColumn, TextColumn, PercentageColumn
console = Console()
def run_simulation():
console.print("[bold cyan]Starting simulated task...[/bold cyan]")
with Progress(
TextColumn("[bold blue]{task.description}[/bold blu…How to Generate a Mock Rollbar Error Report in Python
Create a realistic fake Rollbar error report with random timestamps, levels, messages, and counts for testing and demos.
import json
import random
import time
from datetime import datetime, timedelta
def mock_rollbar_report(n_errors=5):
messages = [
"TypeError: unsupported operand type(s) for +: 'int' and 'str'",
"KeyError: 'user_id'",
"ValueError: invalid literal for int() with base 10: 'abc'",
"At…How to Send and Receive Messages Between Processes with multiprocessing.Pipe in Python
Use multiprocessing.Pipe to create a two-way connection between two processes, send a message from parent to child, and receive a reply back.
import multiprocessing
def child_process(conn):
"""Receive from parent and send back a response."""
message = conn.recv()
print(f"Child received: {message}")
conn.send("Hello from child!")
if __name__ == "__main__":
parent_conn, child_conn = multiprocessing.Pipe()
process = multiprocessing…How to Share a Queue Between Processes in Python
Use multiprocessing.Queue to pass work from a producer process to multiple consumer processes, coordinating with a sentinel stop message.
import multiprocessing
import time
def producer(queue, items):
for item in items:
queue.put(item)
time.sleep(0.1)
queue.put("STOP")
def consumer(queue, name):
while True:
item = queue.get()
if item == "STOP":
break
print(f"{name} processed: {item}")
…How to spawn multiple worker processes in Python with multiprocessing.Process
Spawns three separate worker processes using multiprocessing.Process, runs them concurrently, and waits for all to finish before printing a completion message.
import multiprocessing
import time
def worker(name):
print(f"Worker {name} started")
time.sleep(1)
print(f"Worker {name} finished")
return name
if __name__ == "__main__":
processes = []
for i in range(3):
p = multiprocessing.Process(target=worker, args=(i,))
processes.append(p…How to Assert Exceptions in Python with pytest.raises
Use pytest.raises as a context manager to assert that a function raises an expected exception and inspect its message in pytest tests.
import pytest
def divide(a, b):
if b == 0:
raise ValueError("Cannot divide by zero")
return a / b
def test_divide_by_zero():
with pytest.raises(ValueError) as exc_info:
divide(10, 0)
assert str(exc_info.value) == "Cannot divide by zero"
assert "zero" in str(exc_info.value)
def te…Idempotent Consumer: Store Processed IDs in Python
Implement an idempotent consumer that persists processed message IDs to a JSON file, skipping duplicates on restart.
import json
from pathlib import Path
class IdempotentStore:
def __init__(self, storage_path: str = "processed_ids.json"):
self.storage_path = Path(storage_path)
self.processed_ids = self._load()
def _load(self) -> set:
if self.storage_path.exists():
with self.storage_path…Inbox pattern consumer dedupe mock in Python
Implements a mock inbox consumer that deduplicates incoming messages by ID, with automatic eviction of old seen IDs to prevent unbounded memory growth.
import json
from collections import deque
from dataclasses import dataclass, field
from hashlib import sha256
from typing import Any
@dataclass
class InboxConsumer:
max_seen: int = 1000
seen_ids: set = field(default_factory=set)
seen_history: deque = field(default_factory=deque)
def _mark_seen(self,…Outbox pattern reliable publish in Python with SQLite
Implements a transactional outbox with SQLite, ensuring reliable message publishing by storing events in the same DB transaction as business changes.
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
class Outbox:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS outbox (
id INTEGER PRIMARY KEY AUTO…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.