Reference library

Python Code Samples

Medium snippets you can copy, study, and run in the browser editor.

636 matches
Data pipelines & processing medium

How to Implement SCD Type 1 Overwrite in Python with SQLite

Implement SCD Type 1 dimension updates in Python using SQLite — overwrite existing rows with new data while preserving keys.

scd data-warehouse sqlite
Python
import sqlite3

# Simulate a dimension table with SCD Type 1 (overwrite)
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()

# Create dimension table
cursor.execute("""
    CREATE TABLE customer_dim (
        customer_id INTEGER PRIMARY KEY,
        customer_name TEXT,
        city TEXT,
        updated_at TEXT…
15 0 Open
Data pipelines & processing medium

How to Implement Slowly Changing Dimension Type 2 History in Python

Build a type-2 slowly changing dimension pipeline that closes old records and opens new ones when customer data changes.

scd dimension history
Python
from datetime import datetime, timedelta

def apply_scd_type2(records, current_date):
    """Returns active records after inserting new records with type-2 history."""
    history = []
    active = {}

    for record in records:
        key = record["customer_id"]
        if key in active:
            active[key]["end…
14 0 Open
Data pipelines & processing medium

How to Stream a Large JSONL File Line by Line in Python

Process a large JSON-lines file incrementally using streaming techniques to avoid loading the entire file into memory.

streaming jsonl large-files
Python
import json

def process_large_file(filepath, chunk_size=8192):
    """
    Stream a large JSON-lines file line by line, processing each record
    without loading the entire file into memory.
    """
    total_count = 0
    total_sum = 0
    
    with open(filepath, 'r') as f:
        while True:
            chunk = …
14 0 Open
Data pipelines & processing medium

How to Topologically Sort a DAG in Python

Compute a valid execution order for tasks with dependencies using Kahn's algorithm in Python.

dag topological-sort graph
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…
12 0 Open
Data pipelines & processing medium

How to Validate Fact Table Grain Row Counts in Python

Validate fact table grain by checking dimension key references, unique grain combinations, duplicate rows, and dimension cardinality from a CSV file.

csv data validation etl
Python
import csv
import hashlib
from pathlib import Path


def validate_fact_grain(fact_file: Path, expected_dim_keys: dict[str, set[str]]) -> dict:
    """
    Validate fact table grain by checking each row's dimension keys exist
    in expected dimension tables and row count consistency.
    """
    dim_references = {}
  …
13 0 Open
Data pipelines & processing medium

How to perform a star schema join in Python

Denormalize mock fact and dimension tables by building lookup dicts and enriching each sales fact with customer, product, and date attributes.

star-schema data-joins dimensional-modeling
Python
from datetime import date

# Mock dimension tables
customers = [
    {"customer_id": 1, "name": "Alice", "city": "New York"},
    {"customer_id": 2, "name": "Bob", "city": "Los Angeles"},
    {"customer_id": 3, "name": "Carol", "city": "Chicago"},
]

products = [
    {"product_id": 101, "name": "Laptop", "category": "…
13 0 Open
Data pipelines & processing medium

Implement an Out-of-Order Sort Buffer with a Heap in Python

Buffers out-of-order indices from a stream and emits them in sorted order using a min-heap with a sliding window.

heapq sorting streaming
Python
import heapq
from collections import deque


class OutOfOrderSorter:
    def __init__(self, buffer_size):
        self.buffer_size = buffer_size
        self.buffer = deque(maxlen=buffer_size)
        self.heap = []
        self.next_expected_index = 0
        self.result = []

    def push(self, item):
        heapq.…
12 0 Open
Data pipelines & processing medium

Map Partition Over Chunks in Python with Multiprocessing and Mock

Process data in chunks across multiple CPU cores using multiprocessing Pool.map, and mock the chunk function to test partitioning behavior without heavy computation.

multiprocessing chunking parallel
Python
from multiprocessing import Pool
from unittest.mock import patch, Mock

def process_chunk(chunk):
    return [x * x for x in chunk]

def map_partition_over_chunks(data, chunk_size, process_func=process_chunk):
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with Pool() as pool:
     …
13 0 Open
Data pipelines & processing medium

Normalize Timestamps to UTC DateTime in Python

Convert timestamps in multiple formats to UTC-aware datetime objects using datetime.strptime and astimezone.

datetime timezone utc
Python
from datetime import datetime, timezone

raw_timestamps = [
    "2024-01-15 14:30:00+02:00",
    "17/05/2024 09:15:00 -0500",
    "2024-03-01T22:45:00Z",
    "2024-06-20 08:00:00+09:30"
]

def parse_and_convert(ts: str) -> datetime:
    normalized_ts = ts.strip().replace("Z", "+00:00")
    formats = [
        "%Y-%m-%…
15 0 Open
Data pipelines & processing medium

Pivot long to wide transformation dict

Transform a list of dictionaries from long format to wide format by pivoting on a key column and aggregating values, using pure Python.

pivot transformation data-cleaning
Python
def pivot_long_to_wide(rows, key_col, value_col, id_cols=None):
    """
    Convert long-format data (list of dicts) to wide format.
    
    Args:
        rows: List of dicts in long format
        key_col: Column name to pivot on (becomes new column headers)
        value_col: Column name whose values become the cel…
11 0 Open
Data pipelines & processing medium

Python Exponential Backoff Retry Example

Retry a flaky function with exponential backoff and jitter-free delays, printing each attempt and finally returning the successful result.

retry backoff exception-handling
Python
import random
import time


def flaky_function():
    if random.random() < 0.6:
        raise ConnectionError("Temporary network error")
    return "success"


def retry_with_exponential_backoff(func, max_retries=5, base_delay=1.0):
    for attempt in range(max_retries + 1):
        try:
            return func()
    …
17 0 Open
Git + Python medium

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.

git subprocess repository
Python
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…
13 0 Open
Git + Python medium

Generate CHANGELOG from Conventional Commits in Python

Parse your git log for conventional commits (feat, fix) and produce a simple Markdown CHANGELOG with grouped features and bug fixes.

git changelog automation
Python
import subprocess
import re
import sys
from collections import OrderedDict

CONVENTIONAL_COMMIT = re.compile(
    r"^(?P<type>feat|fix|chore|docs|refactor|perf|test|build|ci|style)(?:\((?P<scope>[^)]+)\))?: (?P<description>.+)"
)


def get_git_log():
    return subprocess.run(
        ["git", "log", "--format=%s"],
  …
15 0 Open
Git + Python medium

Generate Release Notes Markdown from PR Titles in Python

Generate structured Markdown release notes from a list of pull request titles using conventional commit types.

release-notes git pr-titles
Python
import json
from datetime import datetime, timezone

PRS = [
    {"title": "feat: add user login", "number": 12, "merged_at": "2025-01-10"},
    {"title": "fix: resolve payment timeout", "number": 13, "merged_at": "2025-01-11"},
    {"title": "chore: bump dependencies", "number": 14, "merged_at": "2025-01-12"},
    {"…
16 0 Open
Git + Python medium

How to Archive a Repository as a ZIP in Python

Create a ZIP archive of a repository directory with a mock export, skipping hidden files and __pycache__ folders.

zipfile os.walk archiving
Python
import zipfile
import io
import os
from pathlib import Path


def archive_repo_mock(repo_path, output_path="repo_archive.zip"):
    """Create a zip archive of a repository directory (mock export)."""
    repo = Path(repo_path)
    if not repo.exists():
        raise FileNotFoundError(f"Repository not found: {repo}")

…
13 0 Open
Git + Python medium

How to Build a Branch Protection Audit Mock API in Python

A mock HTTP API that serves branch protection rules for repositories and audits them for compliance, built with Python's standard library.

git api http-server
Python
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

REPOSITORIES = {
    "alpha": {
        "default_branch": "main",
        "branches": ["main", "develop", "feature-x"],
        "protection_rules": {
            "main": {"required_reviews": 2, "dismiss_…
12 0 Open
Git + Python medium

How to Format Git Patch Series as an MBOX File in Python

Generate a patch-series mbox file from commit metadata with numbered [PATCH nnn/nnn] subjects and a Git-style footer.

git mbox patch-series
Python
import re
from pathlib import Path


def format_patch_series_mbox(commits, output_path="series.mbox"):
    entries = []
    for idx, commit in enumerate(commits, start=1):
        subject = commit["subject"]
        author = commit["author"]
        email = commit["email"]
        date = commit["date"]
        body = …
15 0 Open
Git + Python medium

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.

git release-notes automation
Python
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,
   …
50 0 Open
Git + Python medium

How to Get Current Git Branch Name in Python with Mock Subprocess

Mocks the subprocess call to reliably test the current git branch name retrieval using GitPython.

git gitpython subprocess
Python
import subprocess
from unittest.mock import patch, MagicMock
from git import Repo
import os


def get_current_branch(repo_path="."):
    """Get the current branch name of a git repository."""
    repo = Repo(repo_path)
    return repo.active_branch.name


if __name__ == "__main__":
    # Mock subprocess to control the…
13 0 Open
Git + Python medium

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.

git logging datetime
Python
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)
  …
13 0 Open
Git + Python medium

How to Mock Git Cherry-Pick in Python for Tests

Mock the `repo.git.cherry_pick` method with `unittest.mock` to test a Git cherry-pick helper without a real repository.

git mock unittest
Python
from unittest.mock import patch, MagicMock

class GitCherryPicker:
    def __init__(self):
        self.applied_commits = []
    
    def cherry_pick(self, commit_hash, repo):
        try:
            result = repo.git.cherry_pick(commit_hash)
            self.applied_commits.append(commit_hash)
            return f"A…
15 0 Open
Git + Python medium

How to Mock Git Pre-commit Hooks (black and ruff) in Python

Mock subprocess to test black and ruff pre-commit commands without actually running them, verifying exit codes.

git pre-commit mocking
Python
import sys
import subprocess
from unittest.mock import patch

def run_hook(command: list[str]) -> int:
    with patch("subprocess.run") as mock_run:
        mock_run.return_value.returncode = 0
        mock_run.return_value.stdout = f"Mocked: {' '.join(command)}"
        result = subprocess.run(command, capture_output…
17 0 Open
Git + Python medium

How to Mock Git Stash and Pop in Python

Mock Git stash, apply, and pop operations using unittest.mock so you can test Git automation without touching a real repository.

git mock gitpython
Python
import git
from unittest.mock import Mock, patch

def stash_and_pop(repo):
    """Mock a stash operation and then pop it back."""
    repo.git.stash("save", "WIP: temp changes")
    stashed_output = repo.git.stash("list")
    
    # Simulate the stash was applied, then pop
    repo.git.stash("apply", "stash@{0}")
    …
15 0 Open
Git + Python medium

How to Mock open() in Python Using unittest.mock.patch

This code shows how to use unittest.mock.patch with mock_open to test a function that checks if a Git patch can be reverse-applied by reading file content.

unittest mocking git
Python
import unittest
from unittest.mock import patch, mock_open


def apply_reverse_check(file_path, expected_patch):
    """
    Check if a patch can be reverse-applied by comparing file content
    with the expected patch's reverse result.
    """
    try:
        with open(file_path, "r") as f:
            content = f.r…
16 0 Open

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

  1. Pick a topic section — strings, lists, files, functions, and more
  2. Open a sample, read How it works, and copy the code block
  3. 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.