Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
Check Null Rate Threshold in PySpark DataFrame
This PySpark code checks the null rate of specified DataFrame columns against a threshold and returns violations.
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, count
def check_null_rate(df, threshold=0.2, columns=None):
"""
Check null rate for specified columns (or all) against a threshold.
Returns columns that exceed the threshold.
"""
cols = columns or df.columns
total…Group Python Events into Sessions with a Gap Timeout
Groups timestamped events into sessions, starting a new session when the time gap exceeds a specified timeout.
from itertools import groupby
from datetime import datetime, timedelta
def session_window_group(events, gap_seconds=300):
"""Group events into sessions where gap > gap_seconds starts a new session."""
if not events:
return []
events = sorted(events, key=lambda x: x[0])
sessions = []
c…How to Filter Data in Python
Filter a list of dictionaries by exact key-value matches or numerical ranges using concise list comprehensions.
from typing import List, Dict, Any
def filter_data(
data: List[Dict[str, Any]], key: str, value: Any
) -> List[Dict[str, Any]]:
"""Return records where data[key] equals value."""
return [record for record in data if record.get(key) == value]
def filter_by_range(
data: List[Dict[str, Any]], key: str…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 Group Rows by Key into Nested Arrays in Python
This code groups rows in a list of dictionaries by a specified key and returns a dictionary with each key mapped to a list of values from another key.
from collections import defaultdict
def implode_rows(rows, key, value_key):
grouped = defaultdict(list)
for row in rows:
grouped[row[key]].append(row[value_key])
return dict(grouped)
if __name__ == "__main__":
data = [
{"category": "fruit", "item": "apple"},
{"category": "fr…How to Implement a Sliding Window Average in Python
Compute the average of the most recent N values in a stream using a bounded deque, efficiently updating the total as new values arrive.
from collections import deque
class SlidingWindowAverage:
def __init__(self, window_size):
self.window_size = window_size
self.window = deque(maxlen=window_size)
self.total = 0
def add(self, value):
if len(self.window) == self.window_size:
self.total -= self.windo…How to Sort a List of Dictionaries by Key in Python
A reusable helper function that sorts a list of dictionaries by a specified key, with optional descending order support.
from typing import List
def sort_records(records: List[dict], key: str, descending: bool = False) -> List[dict]:
"""Sort a list of dictionaries by a specified key."""
return sorted(records, key=lambda record: record[key], reverse=descending)
def demonstrate_sorting() -> None:
users = [
{"name": …How to Topologically Sort a DAG in Python
Compute a valid execution order for tasks with dependencies using Kahn's algorithm in Python.
from collections import defaultdict, deque
def topological_order(dependencies):
graph = defaultdict(list)
in_degree = defaultdict(int)
tasks = set(dependencies.keys())
for task, depends_on in dependencies.items():
for d in depends_on:
graph[d].append(task)
in_degree[t…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 …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")
…Find the Commit That Introduced a String in Git History Using Python
Use git log -S with Python subprocess to find the earliest commit that introduced a specific string across your repository history.
import subprocess
import sys
def find_introducing_commit(repo_path: str, search_string: str, file_glob: str = "*") -> str:
"""Find the first commit that introduced a given string in a git repository."""
result = subprocess.run(
["git", "-C", repo_path, "log", "--all", "--oneline", "-S", search_string…How to Push Git Tags to a Remote with Python
Push specified git tags (or all tags) to a remote repository using Python's subprocess module with error handling.
import subprocess
import sys
def push_tags_to_remote(remote: str = "origin", tags: list[str] | None = None) -> None:
"""
Push git tags to a remote repository.
If no tags are given, push all local tags.
"""
if tags:
subprocess.run(["git", "push", remote, *tags], check=True)
else:
…Merge branch no ff mock in Python
Simulate a Git non-fast-forward merge in Python, producing a synthetic merge commit log for branches with differing SHAs.
class MergeResult:
def __init__(self, base, branch):
self.base = base
self.branch = branch
self.commit_log = []
self.merged = False
def simulate_merge(self):
"""Simulate a 'no-ff' merge by creating a new commit that references both branches."""
if self.base == s…Generate a Mock Presigned URL in Python with HMAC
Build a mock AWS S3 presigned URL using an HMAC-SHA256 signature, mimicking the core SigV4 pattern without cloud SDK dependencies.
import hashlib
import hmac
import time
import base64
def generate_presigned_url_mock(secret_key, bucket, object_key, expires_in=3600):
# Build the canonical request string (simplified AWS SigV4 style)
timestamp = str(int(time.time()))
expiry = str(int(time.time()) + expires_in)
payload = f"GET\n/{buck…How to Calculate Cloud Cost Estimates with a Python Dictionary
Mocks a cloud pricing calculator using a dictionary of service rates and computes total estimated cost for given service hours.
def estimate_cost(service, hours, rate_table=None):
if rate_table is None:
rate_table = {
"basic": 50,
"standard": 75,
"premium": 100
}
if service not in rate_table:
raise ValueError(f"Unknown service: {service}")
return rate_table[service] * hour…How to Calculate VPC Subnet CIDR Details in Python
Compute network address, broadcast address, address count, prefix length, and netmask for any IPv4 CIDR using the Python standard library's ipaddress module.
import ipaddress
def subnet_details(cidr: str) -> dict:
network = ipaddress.ip_network(cidr, strict=False)
return {
"network_address": str(network.network_address),
"broadcast_address": str(network.broadcast_address),
"num_addresses": network.num_addresses,
"prefix_length": ne…How to Check an SCP Deny List in Python
Load a JSON SCP policy file, extract the deny_list, and check if a target ARN is denied.
import json
from pathlib import Path
def evaluate_scp_deny_list(policy_path: Path, target_path: str) -> bool:
policy = json.loads(policy_path.read_text())
deny_list = policy.get("deny_list", [])
return target_path in deny_list
if __name__ == "__main__":
policy_file = Path("scp_policy.json")
pol…How to Enforce Tag Policies on AWS Resources in Python
Build a reusable Python class that checks AWS resources against a required-tag policy and reports compliance with missing tags.
import json
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass
class Resource:
arn: str
tags: Dict[str, str] = field(default_factory=dict)
class TagPolicyEnforcer:
def __init__(self, required_tags: List[str]):
self.required_tags = set(required_tags)
def enfor…How to Evaluate IAM Policy Allow vs Deny in Python
Evaluate an AWS-style IAM policy dict with explicit deny overriding allow and default deny.
import json
def evaluate_policy(action, resource, policy):
"""Evaluate an IAM-like policy dict.
Explicit deny wins over allow. Default is deny.
"""
for statement in policy.get("Statement", []):
effect = statement.get("Effect")
actions = statement.get("Action", [])
resources = …How to Mock AWS Secrets Manager in Python
Create a lightweight mock of AWS Secrets Manager's get_secret_value API to test secret retrieval without cloud dependencies.
import json
from typing import Optional
class MockSecretsManager:
"""A simple mock of AWS Secrets Manager's get_secret_value API."""
def __init__(self):
self._secrets: dict[str, str] = {}
def create_secret(self, secret_id: str, secret_value: str) -> None:
"""Store a secret value under a…How to Mock Auto Scaling Policy Scale Out in Python
Define a mock auto-scaling function that scales out capacity by a factor up to a max, simulating AWS-like events.
def mock_scale_out(current_capacity: int, max_capacity: int, scale_factor: int = 1) -> tuple:
"""
Mock auto-scaling policy: scales out by the specified factor
if capacity allows, capped at max_capacity.
"""
if current_capacity >= max_capacity:
return current_capacity, False
new_cap…How to Mock Azure Service Bus Queue in Python
A lightweight in-memory mock of the Azure Service Bus queue API for local testing without cloud dependencies.
import json
import time
from collections import deque
class ServiceBusQueueMock:
def __init__(self, queue_name):
self.queue_name = queue_name
self._messages = deque()
self._dead_letter_queue = deque()
self._message_counter = 0
def send_message(self, body, message_id=None, prop…How to Mock CloudFront Invalidation Paths in Python
Build a sorted, deduplicated list of CloudFront invalidation paths from a set of file paths, adding implicit index.html entries.
import argparse
def build_invalidation_paths(files, include_index=True):
"""
Create CloudFront invalidation paths from a list of files.
Converts file names to root-relative paths and optionally adds /index.html.
"""
paths = []
for f in files:
f = f.strip()
if not f:
…How to Validate AWS Security Group Ingress Rules in Python
Validates AWS security group ingress rules (protocol, port ranges, CIDR, description) and returns a list of errors or OK.
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class SecurityGroupRule:
protocol: str
port_range: tuple
cidr: str
description: str = ""
def validate_ingress_rule(rule: SecurityGroupRule) -> List[str]:
"""Validate a security group ingress rule against common AWS pat…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.