Reference library

Python Code Samples

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

51 matches
Data pipelines & processing medium

Deduplicate events by ID within a window in Python

Deduplicate event streams by ID within sliding time windows, keeping the newest occurrence per window using heaps and sets.

deduplication events heapq
Python
import heapq
from collections import defaultdict

def deduplicate_events(events, window_size):
    """Return events deduplicated by id, keeping newest within each sliding window."""
    # Index events by (timestamp, id) for deterministic ordering
    events_by_id = defaultdict(list)
    for ts, eid, *payload in events…
14 0 Open
Data pipelines & processing easy

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.

deque sliding-window streaming
Python
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…
15 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
Modern tooling easy

How to Build a Chainable Filter Helper in Python

A beginner-friendly dataclass helper that chains filters, uniqueness, and slicing on any sequence, returning a plain list at the end.

dataclass chaining filter
Python
from dataclasses import dataclass
from typing import Callable, Iterator, Sequence, TypeVar

T = TypeVar("T")


@dataclass
class FilterAssistant:
    """Beginner-friendly helper to filter any collection."""

    data: Sequence[T]

    def where(self, predicate: Callable[[T], bool]) -> "FilterAssistant":
        return …
14 0 Open
System design patterns easy

How to Build an MVP Presenter View Mock in Python

A minimal MVP (Model-View-Presenter) mock showing a Presenter controlling a SlideDeck model with slide navigation and typed state via dataclasses.

dataclasses mvp design-patterns
Python
from dataclasses import dataclass, field
from typing import List


@dataclass
class SlideDeck:
    title: str
    slides: List[str] = field(default_factory=list)
    current_index: int = 0

    def next_slide(self) -> str:
        if self.current_index < len(self.slides) - 1:
            self.current_index += 1
      …
13 0 Open
API design & gRPC easy

How to Implement Pagination with Offset and Limit in Python

A mock API pagination pattern that parses page and per_page query parameters, computes offset and limit, and slices a list of items for a specific page.

api pagination query-params
Python
def paginate(items, page, per_page):
    offset = (page - 1) * per_page
    return items[offset:offset + per_page]


def parse_query_params(query_string):
    params = {}
    if query_string:
        for pair in query_string.split("&"):
            key, value = pair.split("=")
            params[key] = value
    page …
12 0 Open
Streaming & messaging easy

How to Implement a Tumbling Window Counter in Python

Count events that fall within a fixed-size sliding time window using a deque and pruning logic.

streaming window aggregation
Python
from collections import deque
import time


class TumblingWindowCounter:
    def __init__(self, window_size_seconds):
        self.window_size = window_size_seconds
        self.window = deque()

    def add_event(self, timestamp):
        self.window.append(timestamp)

    def count(self, current_time):
        while…
14 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
Streaming & messaging easy

Sliding Window Average with Deque in Python

Computes the running average of a sliding window over streaming numbers using a collections.deque for O(1) pop-left operations.

sliding-window deque streaming
Python
from collections import deque

class SlidingAverage:
    def __init__(self, window_size):
        self.window_size = window_size
        self.window = deque()
        self.total = 0

    def add(self, value):
        self.window.append(value)
        self.total += value
        if len(self.window) > self.window_size:
…
13 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 easy

Build a Rate Limiter Decorator in Python

This code defines a reusable rate limiter decorator that caps function calls within a sliding time window using a deque and monotonic time.

rate-limiting decorator time
Python
import time
from collections import deque


def rate_limiter(max_calls: int, period: float):
    calls = deque()

    def decorator(func):
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            while calls and now - calls[0] >= period:
                calls.popleft()
            if len(ca…
13 0 Open
Reliability & rate limiting easy

How to Build a Rate Limiter in Python

Implements a simple sliding-window rate limiter that caps the number of calls per period, used to throttle processing of a data list.

rate-limiting time sliding-window
Python
import time

class RateLimiter:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.timestamps = []

    def allow(self):
        now = time.time()
        self.timestamps = [t for t in self.timestamps if now - t < self.period]
        if len(self.tim…
14 0 Open
Reliability & rate limiting easy

How to Implement a Sliding Window Counter in Python

This code implements an approximate sliding window counter using a deque of time-based buckets to track event counts within a recent time window.

sliding-window rate-limiting deque
Python
from collections import deque
from time import time


class SlidingWindowCounter:
    def __init__(self, window_size, bucket_size=1):
        self.window_size = window_size
        self.bucket_size = bucket_size
        self.buckets = deque()

    def _evict_expired(self, now):
        while self.buckets and self.buck…
13 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
Reliability & rate limiting medium

How to Implement a Token Bucket Rate Limiter per Client IP in Python

Implements a simple sliding-window rate limiter using a dictionary of timestamp lists per client IP to limit requests per window.

rate-limiting sliding-window ip
Python
from time import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.clients = defaultdict(list)

    def allow(self, ip: str) -> bool:
        now…
13 0 Open
Reliability & rate limiting easy

How to implement rate limiting in Python

Build a simple sliding-window rate limiter in Python that enforces a max number of calls per time period and formats data with timestamps.

rate-limiting time sliding-window
Python
import time

class RateLimiter:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.calls = []
    
    def allow(self):
        now = time.time()
        # Remove calls older than the period window
        self.calls = [t for t in self.calls if now -…
17 0 Open
Reliability & rate limiting medium

How to implement rate limiting per API key in Python

A simple sliding-window rate limiter that tracks request timestamps per API key and rejects requests exceeding the configured limit.

rate-limiting api time-window
Python
import time

API_RATE_LIMITS = {"api_key_1": 5, "api_key_2": 3}  # max requests per window
WINDOW_SECONDS = 10

class RateLimiter:
    def __init__(self, limits, window):
        self.limits = limits
        self.window = window
        self.requests = {key: [] for key in limits}

    def allow(self, api_key):
       …
13 0 Open
Reliability & rate limiting easy

Rate Limit per User ID in Python with a Dict Mock

Implements a simple sliding window rate limiter using a defaultdict of timestamps per user ID, blocking requests that exceed a max count within a time window.

rate-limiting defaultdict sliding-window
Python
import time
from collections import defaultdict


class RateLimiter:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.user_timestamps = defaultdict(list)

    def allow_request(self, user_id):
        now = time.tim…
14 0 Open
Reliability & rate limiting easy

Rate Limiting in Python with a Sliding Window

A beginner-friendly dataclass-based sliding window rate limiter that controls how many calls are allowed per time window.

rate-limiting sliding-window dataclass
Python
import time
from dataclasses import dataclass


@dataclass
class RateLimiter:
    max_calls: int
    window_seconds: float = 1.0

    def __post_init__(self):
        self.calls = []
        self._start = time.monotonic()

    def _update(self, now):
        self.calls = [t for t in self.calls if now - t < self.window…
12 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
Observability & SRE easy

How to Compute SRE Metrics Like Error Rate and Availability in Python

Tracks log events in a sliding time window and calculates error rate per second and availability percentage using an easy-to-follow class.

observability sre metrics
Python
from collections import deque
from datetime import datetime, timedelta
from typing import Dict, Deque


class LogMetrics:
    """Simple observability helper to track log events and calculate SRE metrics."""

    def __init__(self, window_seconds: int = 60):
        self.window_seconds = window_seconds
        self.eve…
14 0 Open
Observability & SRE medium

How to Group Alerts by Time Window in Python

Group alert occurrences that fall within a sliding time window per alert key, reducing noise and summarizing bursts into single events.

alerts grouping monitoring
Python
from collections import defaultdict
from datetime import datetime, timedelta

def group_alerts(alerts, window_minutes=10):
    """Group alerts that occur within the same time window."""
    alerts_by_key = defaultdict(list)
    
    for alert in alerts:
        key = alert["key"]
        timestamp = alert["timestamp"]…
12 0 Open
Observability & SRE easy

How to mock SLI availability success ratio in Python

Simulate request outcomes with deterministic randomness and compute the SLI availability success ratio to check if a target is met.

sli availability monitoring
Python
import random
from collections import defaultdict

def mock_availability(num_requests=1000, target_ratio=0.995):
    """
    Simulate request outcomes and compute the SLI availability success ratio.
    
    Args:
        num_requests: Total number of requests to simulate
        target_ratio: Target availability rati…
14 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

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.