Reference library

Python Code Samples

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

57 matches
Concurrency & performance medium

How to Share Memory Between Processes in Python with multiprocessing.Value and Array

Share a numeric value and a list-like array across multiple Python processes using multiprocessing.Value and multiprocessing.Array, with each process modifying the same memory.

multiprocessing shared-memory concurrency
Python
import multiprocessing

def worker(shared_value, shared_array, index):
    shared_value.value += 10
    shared_array[index] = shared_array[index] * 2

if __name__ == "__main__":
    shared_value = multiprocessing.Value("i", 5)
    shared_array = multiprocessing.Array("i", [1, 2, 3, 4, 5])

    processes = []
    for i…
14 0 Open
Concurrency & performance medium

How to Share a Dict and List Between Processes with multiprocessing Manager in Python

This code demonstrates how to share a dictionary and a list between multiple processes using multiprocessing.Manager, enabling safe concurrent updates.

multiprocessing manager shared-state
Python
import multiprocessing as mp


def worker(shared_dict, shared_list, name):
    shared_dict[name] = name.upper()
    shared_list.append(name)
    print(f"{name} added to shared structures")


def main():
    with mp.Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()

       …
14 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 threading.RLock in Python

Demonstrates threading.RLock, a reentrant lock that allows the same thread to acquire it multiple times without deadlocking — essential for recursive functions sharing state across threads.

threading rlock concurrency
Python
import threading
import time

lock = threading.RLock()
shared_counter = 0

def recursive_increment(value, depth):
    global shared_counter
    with lock:
        shared_counter += 1
        print(f"Depth {depth}: counter = {shared_counter}")
        if depth > 1:
            recursive_increment(value, depth - 1)

def…
14 0 Open
Testing & modern typing medium

How to Mock an Object Method in Python unittest

Mock a method on an instance or class with @patch.object, set its return value, and assert its call arguments in Python unittest.

unittest mock patch
Python
import unittest
from unittest.mock import patch

class Calculator:
    def add(self, a, b):
        return a + b
    
    def multiply(self, a, b):
        return a * b

class TestCalculator(unittest.TestCase):
    def test_add_normal(self):
        calc = Calculator()
        result = calc.add(2, 3)
        self.asse…
14 0 Open
Testing & modern typing medium

How to Use Stubs, Fakes, Spies, and Mocks in Python Testing

Implement four types of test doubles — stubs, fakes, spies, and mocks — as subclasses of a PaymentGateway interface to replace real dependencies during testing.

testing mocks stubs
Python
class PaymentGateway:
    def charge(self, amount):
        raise NotImplementedError


class StubPaymentGateway(PaymentGateway):
    """Returns a fixed response without any logic."""
    def charge(self, amount):
        return {"success": True, "transaction_id": "stub-12345"}


class FakePaymentGateway(PaymentGatewa…
12 0 Open
System design patterns medium

How to Implement CQRS with Separate Read and Write Models in Python

Implements Command Query Responsibility Segregation (CQRS) by splitting data into separate write and read models with dedicated repositories, using dataclasses for structure.

cqrs dataclasses repositories
Python
from dataclasses import dataclass, field
from typing import List, Dict, Optional


@dataclass
class OrderWriteModel:
    order_id: int
    customer: str
    items: List[str] = field(default_factory=list)

    def add_item(self, item: str) -> None:
        self.items.append(item)


@dataclass
class OrderReadModel:
    …
14 0 Open
System design patterns medium

How to Implement the Abstract Factory Pattern in Python

Implements the Abstract Factory pattern to create families of related GUI objects (buttons, checkboxes) without specifying their concrete classes.

abstract-factory design-patterns system-design
Python
from abc import ABC, abstractmethod


class Button(ABC):
    @abstractmethod
    def render(self):
        pass


class Checkbox(ABC):
    @abstractmethod
    def render(self):
        pass


class WindowsButton(Button):
    def render(self):
        return "Rendering Windows-style button"


class WindowsCheckbox(Chec…
13 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
System design patterns medium

Template Method Workflow Steps Base Class in Python

Define a reusable workflow skeleton in a base class and let subclasses fill in each step with the Template Method design pattern.

template-method design-patterns abc
Python
from abc import ABC, abstractmethod


class DataPipeline(ABC):
    """Template Method pattern: defines a workflow skeleton."""

    def run(self):
        """Template method - defines the algorithm's structure."""
        result = {"extracted": False, "transformed": False, "loaded": False}
        raw_data = self._ext…
13 0 Open
API design & gRPC medium

How to mock Server-Sent Events (SSE) in Python

A minimal HTTP server that streams Server-Sent Events to clients, perfect for testing and development.

sse server-sent-events http
Python
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time

MESSAGES = iter([
    "data: Hello world\n\n",
    "data: Second message\n\n",
    "event: custom\n",
    "data: Custom event payload\n\n",
    "data: Final message\n\n"
])

class SSEHandler(BaseHTTPRequestHandler):
    def do_GET…
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
Caching & Redis medium

Cache Penetration Null Object Mock in Python

Implement a cache that stores a null marker on misses to prevent repeated database hits, reducing cache penetration.

caching null-object ttl
Python
import time
from collections import defaultdict
from typing import Any, Optional


class Cache:
    def __init__(self):
        self.store: dict[str, Any] = {}
        self.ttl: dict[str, float] = {}
        self.null_marker = object()

    def get(self, key: str, ttl: int = 60, fallback:
            Any = None) -> An…
17 0 Open
Caching & Redis medium

How to Build a Bloom Filter to Reduce Cache Misses in Python

Implement a probabilistic Bloom filter in Python that lets a cache quickly determine which keys are definitely not present, reducing expensive source lookups on cache misses.

bloom-filter caching probabilistic
Python
import hashlib
import random

class BloomFilter:
    def __init__(self, size=100, num_hashes=3):
        self.size = size
        self.num_hashes = num_hashes
        self.bit_array = [0] * size

    def _hashes(self, item):
        result = []
        for i in range(self.num_hashes):
            hash_value = int(hash…
14 0 Open
Caching & Redis medium

How to Implement a Negative Cache with TTL in Python

This code provides a TTL mock cache that stores negative results (cache misses) for a short time to reduce repeated lookups of missing keys.

cache ttl negative-cache
Python
from time import time, sleep

class TTLMockCache:
    def __init__(self, ttl_seconds=5):
        self.ttl = ttl_seconds
        self.store = {}
        self.negative_cache = {}

    def get(self, key):
        now = time()
        if key in self.store:
            value, expires_at = self.store[key]
            if exp…
13 0 Open
Caching & Redis medium

How to Implement an LFU Cache in Python

Implement a Least Frequently Used (LFU) cache with frequency tracking dictionaries to evict the least accessed items when capacity is reached.

lfu cache frequency
Python
class LFUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.data = {}
        self.freq = {}
        self.min_freq = 0

    def get(self, key: int) -> int:
        if key not in self.data:
            return -1
        self._increment_freq(key)
        return self.data[key]

  …
12 0 Open
Caching & Redis medium

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.

cache redis lru_cache
Python
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…
15 0 Open
Reliability & rate limiting medium

At Least Once with Idempotent Consumer in Python

Implements a thread-safe idempotent consumer that processes each unique message exactly once, even when a producer sends duplicates under an at-least-once delivery model.

idempotency at-least-once threading
Python
import threading
import time
import uuid
from collections import Counter


class IdempotentConsumer:
    def __init__(self):
        self.processed = set()
        self._lock = threading.Lock()

    def consume(self, message_id, payload):
        with self._lock:
            if message_id in self.processed:
          …
16 0 Open
Observability & SRE medium

Export Metrics with OTLP Mock in Python

Simulates system metric collection and exports them as an OTLP-like JSON payload using only Python's standard library.

otlp metrics observability
Python
from dataclasses import dataclass, asdict
import json
import random
import time


@dataclass
class Metric:
    name: str
    value: float
    timestamp: int
    unit: str = "1"


def collect_system_metrics() -> list[Metric]:
    """Mock metric collection for OTLP export simulation."""
    now = int(time.time())
    re…
13 0 Open
Microservices patterns medium

CQRS with Separate Read and Write Repositories in Python

Implement CQRS in Python with separate write and read repositories, using commands for mutations and frozen DTOs for queries.

cqrs repositories microservices
Python
from dataclasses import dataclass
from typing import Dict, List, Optional


# --- Write side: commands mutate state ---
@dataclass
class CreateUserCommand:
    id: int
    name: str


class UserWriteRepository:
    def __init__(self) -> None:
        self._store: Dict[int, Dict[str, object]] = {}

    def create(self,…
15 0 Open
Microservices patterns medium

Consumer Driven Contract Pact Mock in Python

Define and verify consumer-driven contracts using Pact's Consumer and Provider classes, mocking the provider to assert expected interactions.

pact contract testing microservices
Python
from pact import Consumer, Provider

pact = Consumer('OrderService').has_pact_with(Provider('InventoryService'))

@Pact.verify()
class TestInventoryContract:
    def test_get_inventory(self):
        expected = {"item": "widget", "quantity": 100}
        (pact
         .given('inventory exists for widget')
         .u…
17 0 Open
Microservices patterns medium

How to Mock a Choreography Saga in Python

Simulate a choreography-based saga with event envelopes, status tracking, and compensating actions to model distributed transactions.

saga microservices events
Python
import json
from dataclasses import dataclass, asdict
from typing import List, Optional
from enum import Enum


class SagaStatus(Enum):
    PENDING = "PENDING"
    COMPLETING = "COMPLETING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"


@dataclass
class EventEnvelope:
    event_type: str
    order_id: str
    sta…
13 0 Open
Microservices patterns medium

How to implement the Database per service pattern in Python

Simulate separate databases per microservice in Python using dataclasses and in-memory dictionaries, showing how services own their data independently.

microservices database-per-service dataclasses
Python
import json
from dataclasses import dataclass, asdict
from typing import Dict, List


@dataclass
class User:
    id: int
    name: str
    email: str


@dataclass
class Order:
    id: int
    user_id: int
    product: str
    amount: float


class UserServiceDB:
    """Simulates a separate database for the User servic…
12 0 Open
Big data & Spark medium

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.

spark streaming micro-batch
Python
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…
13 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.