Reference library

Python Code Samples

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

11 matches
Comprehensions & generators medium

How to Build a Backpressure Generator Pause Producer Demo in Python

Demonstrates a producer–consumer pattern with a fixed-size buffer that pauses production when full, simulating backpressure.

backpressure producer-consumer deque
Python
import time
import collections

def producer(buffer, max_size, items):
    """Adds items to the buffer until full, then pauses."""
    for item in items:
        while len(buffer) >= max_size:
            print(f"Buffer full ({len(buffer)}/{max_size}) — producer paused")
            time.sleep(0.1)
        buffer.appe…
14 0 Open
Concurrency & performance medium

Benchmark list.append vs deque.append in Python

Measures and compares the performance of appending to a Python list versus a collections.deque using timeit.repeat, showing best and average timings.

benchmark performance list
Python
"""Benchmark list.append vs collections.deque.append."""

import timeit

def bench(stmt, setup, repeat=5, number=1_000_000):
    times = timeit.repeat(stmt, setup=setup, repeat=repeat, number=number)
    return min(times), sum(times) / len(times)

if __name__ == "__main__":
    number = 1_000_000
    list_best, list_a…
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
Streaming & messaging medium

How to Stream Join Windowed Mock Topics in Python

Simulates two message topics and joins their events when timestamps fall within a sliding time window using Python generators and deques.

streaming join generator
Python
import itertools
import random
import time
from collections import deque
from dataclasses import dataclass, field

@dataclass
class Event:
    key: str
    value: int
    timestamp: float = field(default_factory=time.time)

def generate_topic(prefix, keys, start_time):
    while True:
        yield Event(
            …
14 0 Open
Caching & Redis medium

Redis-inspired sliding window rate limiter in Python

A pure-Python sliding window rate limiter using a deque of timestamps, mock-ready for Redis-backed production limits.

redis rate-limit sliding-window
Python
import time
from collections import deque


class SlidingWindowRateLimiter:
    def __init__(self, max_requests: int, window_seconds: int) -> None:
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: dict[str, deque] = {}

    def is_allowed(self, client_id: str…
14 0 Open
Reliability & rate limiting medium

Circuit breaker failure threshold count in Python

Track consecutive or time-windowed failures with a deque to open a circuit breaker and auto-recover to half-open after a cooldown.

circuit-breaker resilience deque
Python
from collections import deque
from time import time, sleep


class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_time: float = 10.0):
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.failures: deque[float] = deque()
        self.st…
16 0 Open
Reliability & rate limiting medium

How to Implement a Sliding Window Log Rate Limiter in Python

Implements a sliding window log rate limiter in Python using a deque of timestamps to enforce a maximum request count within a rolling time window.

rate-limiting sliding-window deque
Python
from collections import deque
from datetime import datetime, timedelta
from time import sleep


class SlidingWindowLog:
    def __init__(self, window_seconds: int, max_requests: int):
        self.window_seconds = window_seconds
        self.max_requests = max_requests
        self.timestamps = deque()

    def allow_…
15 0 Open
Observability & SRE medium

How to Build a Burn Rate Alert with Multiple Time Windows in Python

Track token consumption and trigger alerts when the burn rate exceeds a threshold across multiple time windows using deque and time-based sliding windows.

burn-rate alerts time-windows
Python
import time
from collections import deque

class BurnRateAlert:
    def __init__(self, windows_seconds=(60, 300, 900), threshold_rate=0.8):
        self.windows = {w: deque() for w in windows_seconds}
        self.threshold_rate = threshold_rate
        self.previous_tokens = None

    def record_sample(self, current_…
16 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
Big data & Spark medium

How to Mock and Test a Rate-Limited Source Stream in Python

Build a class that rate-limits emitted items using a sliding window and test it with a simulated stream in Python.

rate-limiting mock-testing streaming
Python
import time
from collections import deque


class RateLimitedSource:
    def __init__(self, max_rate, window=1.0):
        self.max_rate = max_rate
        self.window = window
        self._timestamps = deque()

    def emit(self, item):
        now = time.monotonic()
        while self._timestamps and self._timestam…
16 0 Open
Big data & Spark medium

How to implement a tumbling window aggregation in Python

Build a mock tumbling window aggregator in Python that groups streaming events into fixed time intervals and computes count, sum, and average per window.

tumbling-window streaming aggregation
Python
import time
from collections import deque

class TumblingWindow:
    def __init__(self, duration_seconds):
        self.duration = duration_seconds
        self.buffer = deque()
        self.window_start = None

    def add(self, item):
        current_time = time.time()
        if self.window_start is None:
         …
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.