Reference library

Python Code Samples

Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.

99 matches
Data pipelines & processing easy

Create Data Helper Functions in Python for Beginners

Build reusable Python helper functions to load, filter, sort, summarize, and save JSON data — a beginner-friendly starting point for small data pipelines.

json pipeline helpers
Python
import json
from pathlib import Path
from typing import Any, Dict, List


def load_json_file(filepath: str) -> Dict[str, Any]:
    """Load JSON data from a file."""
    with Path(filepath).open("r", encoding="utf-8") as file:
        return json.load(file)


def filter_by_key(
    data: List[Dict[str, Any]], key: str,…
15 0 Open
Data pipelines & processing easy

How to Clean and Format Data in Python

This code loads JSON data, cleans records by removing empty fields and normalizing text, then summarizes the results with counts and unique keys.

json data cleaning data pipelines
Python
import json
from pathlib import Path


def load_data(filepath: str) -> dict:
    """Load JSON data from a file."""
    with Path(filepath).open("r", encoding="utf-8") as f:
        return json.load(f)


def clean_records(records: list[dict]) -> list[dict]:
    """Remove empty fields and normalize text to lowercase."""…
14 0 Open
Data pipelines & processing easy

How to Parse Data in Python: A Beginner's Helper

This helper parses a JSON payload, extracts user names, emails, and signup dates, then summarizes the results.

json parsing data-processing
Python
import json
from datetime import datetime
from typing import Dict, List


def parse_data(payload: str) -> Dict[str, List]:
    """Parse a JSON payload and extract useful fields."""
    raw = json.loads(payload)
    users = raw.get("users", [])

    parsed = {
        "names": [],
        "emails": [],
        "signup_…
16 0 Open
Data pipelines & processing easy

How to Process CSV Data in Python with a Data Helper

Build a beginner-friendly data helper in Python that loads a CSV file, filters rows by a condition, and summarizes numeric fields.

csv data-processing pathlib
Python
import csv
from pathlib import Path

DATA = [
    {"name": "Alice", "score": 88, "passed": True},
    {"name": "Bob", "score": 42, "passed": False},
    {"name": "Carol", "score": 95, "passed": True},
]


def load_csv(file_path: Path) -> list[dict]:
    with file_path.open(newline="", encoding="utf-8") as f:
        r…
13 0 Open
Cloud + Python medium

Cross Account Role Chaining Mock Credentials in Python

Simulate AWS STS AssumeRole with mock credentials for cross-account role chaining in Python.

aws sts mock
Python
import json

class CredentialChain:
    def __init__(self, account_id, role_name):
        self.account_id = account_id
        self.role_name = role_name
        self.credentials = {}

    def assume_role(self, session_name="mock_session"):
        """Simulate STS AssumeRole, returning mock credentials with expiry.""…
17 0 Open
Cloud + Python easy

How to Create a Mock STS AssumeRole Credentials Dict in Python

Build a realistic AWS STS AssumeRole response dict with temporary credentials, expiry time, and assumed role ARN for local testing.

aws sts mocking
Python
import json
from datetime import datetime, timedelta, timezone


def mock_sts_credentials(role_arn, session_name, duration=3600):
    now = datetime.now(timezone.utc)
    expiration = now + timedelta(seconds=duration)

    credentials = {
        "Credentials": {
            "AccessKeyId": "ASIAEXAMPLEACCESSKEY",
    …
14 0 Open
Concurrency & performance medium

Build a Python Performance Profiler That Generates Readable Reports

Use cProfile and pstats to profile Python functions and print a sorted performance report showing the top time-consuming calls.

profiling cprofile pstats
Python
import cProfile
import pstats
import io
from pathlib import Path

def slow_function():
    total = 0
    for i in range(500_000):
        total += i ** 2
    return total

def fast_function():
    total = sum(i * i for i in range(500_000))
    return total

def profile_functions():
    profiler = cProfile.Profile()
  …
46 0 Open
Concurrency & performance medium

How to Build a Producer-Consumer Pattern with asyncio.Queue in Python

This code implements a classic producer-consumer pattern using asyncio.Queue to coordinate one producer task that generates items and two consumer tasks that process them concurrently, with a sentinel value to signal completion.

asyncio queue concurrency
Python
import asyncio
import random


async def producer(queue, item_count):
    for i in range(item_count):
        item = random.randint(1, 100)
        await queue.put(item)
        print(f"Produced: {item}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # Sentinel to signal end


async def consumer(queue, n…
16 0 Open
Concurrency & performance medium

How to Pause and Resume Threads with threading.Event in Python

Use threading.Event to pause and resume worker threads in Python, controlling execution flow with set and clear methods.

threading events concurrency
Python
import threading
import time

workers = []

def worker(name, event):
    for i in range(10):
        event.wait()
        print(f"{name} step {i}")
        time.sleep(0.1)

def pause_worker(name):
    global pause_event
    for w in workers:
        if w.name == name:
            pause_event.clear()
            print(…
11 0 Open
Concurrency & performance medium

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.

multiprocessing queue concurrency
Python
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}")

…
14 0 Open
Concurrency & performance medium

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.

threading condition producer-consumer
Python
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:
       …
14 0 Open
Concurrency & performance easy

Thread-Safe Producer Consumer Queue in Python

A producer-consumer pattern using thread-safe queue.Queue with two threads, demonstrating safe communication and synchronized task completion.

queue threading producer-consumer
Python
import queue
import threading
import time
import random


def producer(q, item_count):
    for i in range(item_count):
        item = random.randint(1, 100)
        q.put(item)
        print(f"Producer added: {item}")
        time.sleep(0.1)


def consumer(q):
    while True:
        try:
            item = q.get(time…
12 0 Open
Testing & modern typing easy

How to Group Data by Key in Python with Type Hints

Group a list of dictionaries by a specified key using a typed helper function and print a summary of each group.

grouping type-hints dictionaries
Python
from typing import Any, Dict, List, TypeVar, Union

T = TypeVar("T")

def group_by(data: List[Dict[str, Any]], key: str) -> Dict[Any, List[Dict[str, Any]]]:
    """Group a list of dictionaries by a given key."""
    grouped: Dict[Any, List[Dict[str, Any]]] = {}
    for item in data:
        value = item.get(key)
     …
12 0 Open
Testing & modern typing easy

How to Test Hypotheses with Property-Based Check in Python

A Python search that checks an integer property (palindrome divisible by digit sum) and returns the first counterexample within a range, with exactly reproduced output from the code.

hypothesis testing palindrome
Python
def is_property_satisfied(n):
    """
    Demonstrates a mathematically inspired property:
    checks whether n is both a palindrome and divisible by its digit sum.
    """
    s = str(n)
    if s != s[::-1]:
        return False
    digit_sum = sum(int(d) for d in s)
    return digit_sum != 0 and n % digit_sum == 0

…
10 0 Open
Testing & modern typing easy

How to Write a Contract Test with Mock in Python

Use unittest.mock to verify a consumer's expectations match the provider's response shape in a Python contract test.

contract-testing unittest mock
Python
from unittest.mock import Mock

# Contract test: verify consumer expects data shape that provider delivers.
# We mock the provider and assert the consumer's calls match the agreed contract.

def fetch_user(provider_client, user_id):
    """Consumer code: expects provider to return {'id', 'name', 'email'}."""
    respo…
13 0 Open
System design patterns easy

How to Implement a Simple Event Bus in Python

Create a publish-subscribe event bus using dataclasses and defaultdict to decouple event producers from consumers.

event-bus publish-subscribe design-patterns
Python
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Set


@dataclass
class EventBus:
    _subscribers: Dict[str, List[Callable]] = field(
        default_factory=lambda: defaultdict(list)
    )

    def subscribe(self, event_type: str, handler: Callable…
15 0 Open
System design patterns easy

Idempotent Consumer: Store Processed IDs in Python

Implement an idempotent consumer that persists processed message IDs to a JSON file, skipping duplicates on restart.

idempotency duplicate-detection state-persistence
Python
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…
15 0 Open
System design patterns medium

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.

deduplication inbox-pattern dataclasses
Python
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,…
13 0 Open
API design & gRPC easy

How to Build a Data Helper Class in Python for Beginners

Create a beginner-friendly DataHelper class that stores, retrieves, filters, and summarizes records in a list of dictionaries.

dataclasses data-handling beginner
Python
from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


@dataclass
class DataHelper:
    """A beginner-friendly helper for common data tasks."""

    data: List[Dict[str, Any]] = field(default_factory=list)

    def add_record(self, record…
14 0 Open
Streaming & messaging medium

Batch Consume Process Commit Pattern in Python

A mock batch processor that accumulates items in a queue, processes full batches, commits successful or failed results, and flushes remaining items.

streaming batch-processing queues
Python
import random
import threading
import time
from collections import deque


class MockBatchProcessor:
    def __init__(self, process_func, commit_func, batch_size=5):
        self.queue = deque()
        self.batch_size = batch_size
        self.process_func = process_func
        self.commit_func = commit_func

    de…
15 0 Open
Streaming & messaging easy

Exactly Once Idempotent Consumer Store in Python

A mock key-value store that guarantees exactly-once processing by rejecting duplicate message keys in a message or event stream.

idempotency streaming deduplication
Python
from collections import defaultdict

class ExactlyOnceStore:
    def __init__(self):
        self.processed = defaultdict(set)
        self.data = {}

    def consume(self, key, value):
        if key in self.data:
            return False
        self.data[key] = value
        return True

    def get_processed_count…
15 0 Open
Streaming & messaging easy

How to Build a Materialized View Updater Consumer Mock in Python

A mock consumer that queues change events and triggers refresh callbacks to simulate materialized view updates.

dataclasses deque mocking
Python
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Optional


@dataclass
class MaterializedViewUpdater:
    """Mock updater that consumes change events and refreshes a view."""
    refresh: Optional[Callable[[str], None]] = None
    queue: Deque[tuple…
14 0 Open
Streaming & messaging medium

How to Implement Backpressure Pause Producer with a Bounded Queue in Python

Places a Producer thread that sends items into a bounded queue with backpressure: on Full, it pauses to let the consumer catch up.

queues backpressure threading
Python
import threading
import time
import queue
import random


class Producer:
    def __init__(self, q):
        self.q = q
        self.running = True

    def produce(self):
        while self.running:
            item = random.randint(1, 100)
            try:
                self.q.put(item, timeout=0.5)
              …
14 0 Open
Streaming & messaging easy

How to Mock Kafka Topic Partitions with a Python dict of lists

Mocks a Kafka topic and its partitions using a defaultdict of lists to simulate message production, consumption, and per-partition counts.

kafka mock partitions
Python
from collections import defaultdict

class KafkaTopicPartitionMock:
    """A simple mock for Kafka topic-partition assignment using dict of lists."""

    def __init__(self, topic):
        self.topic = topic
        self.partitions = defaultdict(list)  # partition_id -> list of messages

    def produce(self, message…
15 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.