Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
How to Make a Git Commit Heatmap by Hour in Python
Parse a git log output and count commits by weekday and hour, then print a compact heatmap table.
import re
from collections import Counter
from datetime import datetime
def parse_commits(log_text):
"""Parse git log lines and count commits by (weekday, hour)."""
pattern = re.compile(r"^Date:\s+(.+)$")
counts = Counter()
for line in log_text.splitlines():
match = pattern.match(line)
…
How to mock boto3 S3 upload file wrapper in Python
Wrap an S3 put_object call in a testable function that returns metadata, and mock boto3 to verify the upload without touching AWS.
import boto3
import io
def upload_file_to_s3(file_obj, bucket, key, object_metadata=None):
"""Upload a file-like object to S3 and return a metadata dict."""
s3 = boto3.client("s3")
content = file_obj.read()
s3.put_object(
Bucket=bucket,
Key=key,
Body=content,
Metadata=…
Mock CDK Synth Output in Python for Template Testing
Simulate AWS CDK synth output with MagicMock to test or preview CloudFormation templates without running a real CDK app.
import json
from unittest.mock import MagicMock
def mock_cdk_synth() -> dict:
"""Simulate AWS CDK synth output for a simple S3 bucket."""
cdk_app = MagicMock()
cdk_app.synth.return_value.template = {
"Resources": {
"MyBucket": {
"Type": "AWS::S3::Bucket",
…
How to Mock CLI Output in Typer with unittest.mock
Mock and capture Typer CLI output using unittest.mock.patch and io.StringIO for testing command-line applications.
import typer
from unittest.mock import patch
import io
app = typer.Typer()
@app.command()
def greet(name: str, age: int = 18, uppercase: bool = False):
"""Greet a person with optional formatting."""
message = f"Hello {name}, age {age}"
if uppercase:
message = message.upper()
typer.echo(messag…
How to Use ProcessPoolExecutor for CPU Parallel Map in Python
Run a function over a sequence of inputs in parallel across multiple CPU cores with ProcessPoolExecutor.map.
from concurrent.futures import ProcessPoolExecutor
import math
def compute_square(num):
return num * num
def is_prime(n):
if n < 2:
return False
for i in range(2, int(math.sqrt(n)) + 1):
if n % i == 0:
return False
return True
if __name__ == "__main__":
numbers = rang…
How to Use a Bounded Buffer with threading.Condition in Python
Implement a thread-safe bounded buffer using threading.Condition and show a producer–consumer example with exact output.
import threading
import time
import random
class BoundedBuffer:
def __init__(self, capacity):
self.capacity = capacity
self.buffer = []
self.condition = threading.Condition()
def put(self, item):
with self.condition:
while len(self.buffer) >= self.capacity:
…
Profile Memory Usage with tracemalloc Snapshot Diff in Python
Use tracemalloc to take two memory snapshots, compute a diff, and print the top changes (size and count) by line number.
import tracemalloc
def profile_memory():
tracemalloc.start()
# Allocate some objects to track
data = [i * 2 for i in range(10000)]
text = "x" * 5000
nested = {"key": [1, 2, 3], "value": (4, 5)}
# Take first snapshot
snapshot1 = tracemalloc.take_snapshot()
# Free some mem…
Characterization Test for Legacy Python Code
Capture the exact output of a legacy Python function for known inputs, creating a characterization test that documents current behavior before refactoring.
def legacy_behavior(value):
"""Legacy function that returns a tuple with unconventional types."""
if value == "special":
return None, "legacy-special"
elif value > 100:
return value, "large"
elif value > 0:
return value * 2, "positive-doubled"
elif value == 0:
…
How to Run Test Coverage with pytest-cov in Python
Run pytest with coverage reporting using pytest-cov on a temporary project and see line-by-line coverage output.
import os
import subprocess
import tempfile
from pathlib import Path
def sample_function(x: int) -> int:
"""A simple function to demonstrate coverage."""
if x > 0:
return x * 2
else:
return -x
def run_pytest_with_coverage() -> str:
"""Run pytest with coverage on a temp project and r…
How to Snapshot Test JSON with Mock in Python
Use pytest-snapshot to capture the exact output of a JSON-loading function, with and without mocking json.loads, so future changes are automatically detected.
import json
from unittest.mock import Mock, patch
import pytest
def load_config(data):
config = json.loads(data)
return {"host": config["host"], "port": config["port"]}
def test_load_config_snapshot(snapshot):
mock_data = json.dumps({"host": "localhost", "port": 8080, "extra": "ignored"})
result = …
How to Test Properties with Random Inputs in Python
Write a simple property-based test in Python using random string generation to verify that string invariants like reverse-twice identity and uppercase idempotence always hold.
import random
import string
def generate_random_string(length: int) -> str:
"""Generate a random alphanumeric string of given length."""
chars = string.ascii_letters + string.digits
return "".join(random.choice(chars) for _ in range(length))
def reverse_twice_is_identity(s: str) -> bool:
"""Propert…
How to Use TypedDict for Data Validation in Python
Define a TypedDict schema and validate raw dictionary input with type hints for safer, more readable data handling.
from typing import Any, Dict, List, Optional, Union, TypedDict, Literal
class Product(TypedDict):
product_id: int
name: str
price: Union[int, float]
in_stock: bool
tags: Optional[List[str]]
def validate_product(data: Dict[str, Any]) -> Product:
product_id: int = int(data["product_id"])
na…
How to Structure a Three-Tier Layered Architecture in Python
A mock three-tier architecture with presentation, business, and data layers that process a user request from input to response.
class PresentationLayer:
def __init__(self, business_layer):
self.business = business_layer
def handle_request(self, user_id):
print(f"[Presentation] Received request for user {user_id}")
data = self.business.process_user(user_id)
print(f"[Presentation] Response: {data}")
…
Build a Mock REST API with PUT and GET in Python
A minimal mock REST server implementing idempotent PUT for resource replacement and GET for retrieval, built with Python's http.server module.
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
from urllib.parse import urlparse
mock_db = {}
class MockAPIHandler(BaseHTTPRequestHandler):
def do_PUT(self):
parsed = urlparse(self.path)
resource_id = parsed.path.strip("/").split("/")[-1]
content_length = int(self.…
How to Aggregate Periodic Snapshot Data in Python
Generates mock snapshot data and groups values into periods to compute average aggregates with Python's standard library.
import random
from collections import defaultdict
def snapshot_aggregate(n=10, period=3):
data = defaultdict(list)
for i in range(n):
key = f"item_{i % period}"
data[key].append(random.randint(1, 100))
return dict(data)
def aggregate_periodic(snapshots, period=3):
result = {}
for …
Mock Watermark Late Event Side Output in Python
Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.
from datetime import datetime, timedelta
from typing import List, Tuple
def watermark_mock(
events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
"""Simulate watermarking: events arriving on time vs. late by ch…
Cache Stampede Prevention with SingleFlight in Python
Implements a SingleFlight pattern in Python to deduplicate concurrent cache-miss computations and prevent cache stampede.
import threading
import time
from functools import wraps
class SingleFlight:
def __init__(self):
self._lock = threading.Lock()
self._inflight = None
def do(self, key, fn):
with self._lock:
if self._inflight is not None:
return self._inflight[1]
…
How to Implement a Write-Through Cache in Python with a Mock Database
A thread-safe write-through cache that updates both cache and mock database atomically, computing values only after a successful write to the database.
import threading
import time
import random
class WriteThroughCache:
def __init__(self):
self.cache = {}
self.db = {}
self.lock = threading.Lock()
def write(self, key, value):
with self.lock:
# Simulate slow database write
time.sleep(random.uniform(0.01…
Implement a Multi-Level Cache with L1 Memory and L2 Redis in Python
This code implements a simple multi-level cache with an in-process L1 cache (via functools.lru_cache) and a mock Redis L2 cache with TTL, falling back to a slow computation on misses.
import time
from functools import lru_cache
class MockRedis:
def __init__(self):
self.store = {}
def get(self, key):
return self.store.get(key, None)
def set(self, key, value, ttl=5):
self.store[key] = (value, time.time() + ttl)
def get_ttl(self, key):
value, expiry…
How to Build a Python Latency Histogram with Mock Buckets
This code implements a mock latency histogram that records request durations into configurable buckets and outputs counts, total, and average latency.
import time
import random
from collections import Counter
class LatencyHistogram:
def __init__(self, buckets):
self.buckets = sorted(buckets)
self.counts = Counter()
self.total = 0
self.sum_latency = 0
def record(self, latency_ms):
for i, boundary in enumerate(self.bu…
How to Track Cache Hit Ratio in Python
Simulate an LRU cache with hit/miss tracking and compute a real-time hit ratio from random access patterns.
import random
import time
from collections import OrderedDict
class LRUCache:
def __init__(self, capacity: int):
self.cache = OrderedDict()
self.capacity = capacity
self.hits = 0
self.misses = 0
def get(self, key):
if key in self.cache:
self.hits += 1
…
How to Build a DAG Execution Stage Calculator in Python
Computes the execution stages of a directed acyclic graph (DAG) by grouping nodes that become ready simultaneously using topological sorting with Kahn's algorithm.
from collections import defaultdict, deque
def get_stages(edges):
"""Return list of stages, where each stage is a list of nodes
that become ready at the same time in a DAG."""
graph = defaultdict(list)
in_degree = defaultdict(int)
nodes = set()
for src, dst in edges:
graph[src].appen…
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 Mock a Compute-Collect Action Trigger in Python
Mock a compute-collect action trigger using Python's unittest.mock to simulate Spark-style job execution and assert trigger behavior.
Here's a Python code sample for the problem title "Action trigger compute collect mock":
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.