Reference library

Python Code Samples

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

12 matches
OOP & classes medium

How to Create a Data Splitter Class in Python

This code defines a DataSplitter class that splits data by index, into chunks, or by a predicate, demonstrating OOP principles in Python.

class data-splitting slicing
Python
class DataSplitter:
    def __init__(self, data):
        self.data = list(data)
    
    def split_by_index(self, index):
        return self.data[:index], self.data[index:]
    
    def split_into_chunks(self, chunk_size):
        return [self.data[i:i + chunk_size] for i in range(0, len(self.data), chunk_size)]
   …
14 0 Open
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 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
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

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 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
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 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
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
Auth & security at scale medium

How to mock DNS CAA record lookups in Python

Parse and filter DNS CAA records with a mock lookup function, demonstrating how certificate authorities validate domain authorization.

dns security caa
Python
import dnslib

def parse_caa_record(record_string):
    """Parse a DNS CAA record string into its components."""
    parts = record_string.split()
    flags = int(parts[0])
    tag = parts[1]
    value = parts[2]
    return flags, tag, value

def mock_caa_lookup(domain, caa_records):
    """Mock DNS CAA lookup that re…
18 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.