Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Process System Metrics (RSS, CPU) in Python
Simulate and aggregate RSS and CPU system metrics to compute averages and maximums for monitoring dashboards.
import random
import time
from collections import namedtuple
Metric = namedtuple("Metric", ["name", "value", "unit"])
def generate_metrics(num_metrics: int = 5) -> list:
"""Simulate a batch of system metrics."""
metrics = []
for i in range(num_metrics):
rss = random.randint(50, 500) # MB
…How to Simulate a Queue Depth Gauge in Python
Simulate a queue depth over time using a random enqueue/dequeue process, returning depth values that can be used for monitoring or testing dashboards.
import collections
import random
import time
def simulate_queue_depth(max_depth=10, steps=20):
queue = collections.deque()
depth_history = []
for _ in range(steps):
# Randomly enqueue or dequeue
if random.random() < 0.6 and len(queue) < max_depth:
queue.append("task")
…Distributed tracing with contextvars in Python
Propagate trace and span IDs across function calls using contextvars to mock distributed tracing in a single process.
import contextvars
import uuid
import time
_trace_context = contextvars.ContextVar("trace_context", default=None)
class TraceContext:
def __init__(self, trace_id, parent_span_id):
self.trace_id = trace_id
self.parent_span_id = parent_span_id
self.span_id = uuid.uuid4().hex[:16]
s…How to Check an External Gateway vs Use an Internal Mock in Python
This code checks whether an external network gateway is reachable using ping, then falls back to a deterministic internal mock for testing environments.
import subprocess
import sys
def check_external_gateway():
"""True if we can reach an external network target."""
try:
subprocess.run(
["ping", "-c", "1", "-W", "2", "8.8.8.8"],
capture_output=True,
timeout=3,
check=True,
)
return True
…How to Deduplicate Events in Python with SHA256 Hashing
Build an event deduplicator that identifies duplicate inbox messages using SHA256 hashes and tracks duplicate counts per event type.
```python
import hashlib
import json
from collections import defaultdict
class EventDeduplicator:
def __init__(self):
self.seen_hashes = set()
self.duplicate_counts = defaultdict(int)
def process_event(self, event):
event_key = f"{event['event_id']}:{event['timestamp']}"
even…Idempotent Consumer Event Processing in Python
Track processed event IDs to skip duplicates and count event types for a reliable, idempotent consumer.
import json
from collections import defaultdict
class EventProcessor:
def __init__(self):
self.processed_ids = set()
self.counts = defaultdict(int)
def process_event(self, event):
event_id = event["id"]
if event_id in self.processed_ids:
return {"status": "skipped"…How to Implement a Streaming Watermark in Python
Mock structured streaming watermarks in Python to track late event times and compute a watermark for windowed processing.
from datetime import datetime, timedelta
import time
class StreamingWatermark:
"""Mock watermark tracker for structured streaming."""
def __init__(self, watermark_delay_seconds):
self.watermark_delay = timedelta(seconds=watermark_delay_seconds)
self.max_event_time = None
def observe_even…How to Implement row_number Window Function in Python
This code implements a SQL-style ROW_NUMBER() window function in pure Python, partitioning rows by a set of columns and ranking them within each partition by an ordered set of columns.
from collections import defaultdict
import itertools
def row_number(rows, partition_by, order_by):
partitions = defaultdict(list)
for index, row in enumerate(rows):
key = tuple(row[col] for col in partition_by)
partitions[key].append((index, row))
result = []
for key in partitions:
…How to Mock Spark Streaming Micro-Batches in Python
Simulate Spark's micro-batch streaming with a simple deque-based class that collects events over time and processes them in timed batches.
import time
from collections import deque
from datetime import datetime
class MicroBatchStream:
def __init__(self, batch_interval_sec=2):
self.batch_interval = batch_interval_sec
self.source = deque()
self.processed = []
def add_events(self, events):
self.source.extend(events…Bayesian Optimization in Python: A Simplified Mock Implementation
A toy Bayesian optimization loop with a Gaussian process prior, expected improvement acquisition, and noisy sampling to find a function's minimum.
import random
import math
class BayesianOptimizer:
def __init__(self, noise=0.1):
self.noise = noise
self.observations = []
def objective(self, x):
return (math.sin(3*x) + 0.5*x) / (1 + x**2)
def gaussian_process_prior(self, x1, x2, length_scale=0.5):
return math.…How to Build a Mock ML Pipeline with Prefect in Python
Create a lightweight Prefect flow with mock preprocessing, training, and evaluation tasks to prototype an ML pipeline end-to-end.
from prefect import task, flow
from datetime import datetime
@task
def preprocess_data(raw_value: float) -> float:
"""Mock preprocessing: normalize the input value."""
return raw_value / 100.0
@task
def train_model(features: float) -> dict:
"""Mock training: return a fake model artifact."""
return …How to Build an sklearn Pipeline with ColumnTransformer in Python
A mock example showing how to chain preprocessing and a regression model into a single sklearn Pipeline, scaling numeric features and one-hot encoding categorical features with ColumnTransformer.
import numpy as np
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LinearRegression
# Mock dataset
X = np.array([[1, 'red'], [2, 'blue'], [3, 'red'], [4, 'green'], [5, 'blue']], dtype=o…How to Load, Save, and Split JSON Data in Python
Provides helper functions to load, save, and split JSON dictionary data for simple ML pipeline preprocessing.
import json
from pathlib import Path
def load_json_data(file_path):
"""Load JSON data from a file, returning an empty dict if missing."""
path = Path(file_path)
if path.exists():
with path.open("r", encoding="utf-8") as f:
return json.load(f)
return {}
def save_json_data(data, f…How to do feature selection with VarianceThreshold in Python
This code demonstrates how to use scikit-learn's VarianceThreshold to remove low-variance features from a NumPy array, keeping only those that vary enough to be useful for modeling.
import numpy as np
from sklearn.feature_selection import VarianceThreshold
def main():
# Mock dataset: 4 samples, 5 features
X = np.array([
[0.1, 0.2, 1.0, 1.0, 0.5],
[0.2, 0.2, 0.0, 1.0, 0.4],
[0.1, 0.2, 1.0, 1.0, 0.6],
[0.3, 0.2, 1.0, 0.0, 0.5]
])
# Select features w…One Hot Encode Categories in Python
Convert a list of categorical strings into one-hot encoded numeric vectors using pure Python and NumPy.
import numpy as np
categories = ["red", "green", "blue", "red", "blue", "green", "red"]
unique = sorted(set(categories))
lookup = {cat: i for i, cat in enumerate(unique)}
one_hot = []
for cat in categories:
row = [0] * len(unique)
row[lookup[cat]] = 1
one_hot.append(row)
print("Categories:", categories…StandardScaler mock in Python
A pure-Python StandarScaler class that standardizes features to zero mean and unit variance without sklearn.
import math
class StandardScaler:
def __init__(self):
self.mean_ = None
self.std_ = None
def fit(self, X):
n = len(X)
self.mean_ = [sum(col) / n for col in zip(*X)]
self.std_ = []
for col in zip(*X):
variance = sum((x - self.mean_[i]) ** 2 for i, x …Build a Full Text Search Index in Python
Create a simple inverted index for full-text search with the standard library, supporting multi-word AND queries across documents.
import re
from collections import defaultdict
class SimpleTextIndex:
def __init__(self):
self.index = defaultdict(list)
self.documents = {}
def add_document(self, doc_id, text):
self.documents[doc_id] = text
words = set(re.findall(r'\w+', text.lower()))
for word in wo…How to Limit a Result Set to Top N Rows in Python
Sort a list of dictionaries by a numeric key and return only the top N results, formatted as a readable ranked list.
import random
def top_n_mock(limit: int = 5):
"""Return a formatted top-N result set as a mock example."""
# Simulated data source
scores = [
{"name": "Alice", "score": 87},
{"name": "Bob", "score": 92},
{"name": "Charlie", "score": 78},
{"name": "Diana", "score": 95},
…Docker healthcheck CMD mock in Python
Runs a subprocess to curl a health endpoint and returns exit code 0 when healthy, 1 when unhealthy, mimicking a Docker HEALTHCHECK command.
import subprocess
import sys
def run_healthcheck() -> int:
result = subprocess.run(["curl", "-fsS", "http://localhost:8080/health"], capture_output=True, text=True)
if result.returncode == 0:
print("healthy")
return 0
print("unhealthy", file=sys.stderr)
return 1
if __name__ == "__ma…How to Build a Data Helper for Production Deployment in Python
Build a reusable DataHelper class that loads configs, validates required keys, normalizes string values, and logs schema details — a production-ready data processing pattern.
import json
from pathlib import Path
from typing import Any, Dict
class DataHelper:
"""Common data processing patterns for production deployment."""
def __init__(self, config_path: str | Path):
self.config_path = Path(config_path)
self.config = self._load_config()
def _load_confi…How to Mock a Dockerfile Multi-Stage Build in Python
Simulate a Dockerfile multi-stage build process in Python using dataclasses to validate stage ordering and file availability before you write the real Dockerfile.
from dataclasses import dataclass
from pathlib import Path
@dataclass
class BuildStage:
name: str
base_image: str
files: list[str]
commands: list[str]
def run_build(stage: BuildStage, context_dir: Path):
print(f"=== Stage: {stage.name} (base: {stage.base_image}) ===")
for file in stage.file…How to Mock a SIGTERM Handler in Python
Create a graceful shutdown handler for SIGTERM and SIGINT signals, then test it by simulating a signal delivery without terminating the process.
import signal
import time
class Service:
def __init__(self):
self.running = True
def shutdown(self, signum, frame):
print(f"Received signal {signum}, shutting down gracefully...")
self.running = False
def run(self):
signal.signal(signal.SIGTERM, self.shutdown)
sig…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.