Python Code
Samples
Copy-ready Python snippets by topic and difficulty — short, focused, and runnable in the browser editor.
How to Record Last N Errors with a Ring Buffer in Python
Use collections.deque with maxlen to keep only the most recent N error messages while discarding older entries automatically.
import collections
class ErrorRecorder:
def __init__(self, size):
self.buffer = collections.deque(maxlen=size)
def record_error(self, message):
self.buffer.append(message)
def get_errors(self):
return list(self.buffer)
if __name__ == "__main__":
recorder = ErrorRecorder(3)
…How to Implement a Queue Class in Python Using deque
Build a FIFO queue class in Python backed by the collections.deque container with enqueue, dequeue, peek, and size methods.
from collections import deque
class Queue:
def __init__(self):
self._items = deque()
def enqueue(self, item):
self._items.append(item)
def dequeue(self):
if self.is_empty():
raise IndexError("dequeue from empty queue")
return self._items.popleft()
…How to Get the Breadth-First Traversal Order of a Graph in Python
Performs a breadth-first search on an adjacency list and returns the order nodes are visited, using a deque for efficient queue operations.
from collections import deque
def bfs_order(adjacency, start=0):
"""Return the order nodes are visited in a breadth-first traversal."""
visited = set()
order = []
queue = deque([start])
visited.add(start)
while queue:
node = queue.popleft()
order.append(node)
for neig…How to Implement a Moving Average from a Data Stream in Python
Implement a MovingAverage class using a deque and running sum to compute the average of the last k values from a continuous data stream.
from collections import deque
class MovingAverage:
def __init__(self, size):
self.size = size
self.queue = deque()
self.window_sum = 0
def next(self, val):
self.queue.append(val)
self.window_sum += val
if len(self.queue) > self.size:
self.window_su…How to Implement a Recent Counter with a Deque in Python
Implements a RecentCounter class that uses a deque to count ping requests within the last 3000 milliseconds.
from collections import deque
import time
class RecentCounter:
def __init__(self):
self.hits = deque()
def ping(self, t: int) -> int:
self.hits.append(t)
while self.hits and self.hits[0] < t - 3000:
self.hits.popleft()
return len(self.hits)
if __name__ == "__mai…Implement Queue Using Two Stacks in Python
Python class that implements a FIFO queue using two stacks, with enqueue, dequeue, peek, and emptiness checks.
class QueueUsingStacks:
def __init__(self):
self.stack_in = []
self.stack_out = []
def enqueue(self, value):
self.stack_in.append(value)
def dequeue(self):
if not self.stack_out:
while self.stack_in:
self.stack_out.append(self.stack_in.pop())
…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.
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…How to Keep Last K Turns in a Memory Buffer in Python
A TurnBuffer class using deque with maxlen to keep only the most recent k conversation turns in memory for LLM context.
from collections import deque
class TurnBuffer:
def __init__(self, k):
self.k = k
self.turns = deque(maxlen=k)
def add(self, turn):
self.turns.append(turn)
def last_k(self):
return list(self.turns)
if __name__ == "__main__":
buffer = TurnBuffer(3)
buffer.add("tu…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.
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…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 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…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.
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,…Build a Streaming Messaging Helper in Python
Create a simple message stream class that stores recent messages, sends user messages, and retrieves history or latest messages with timestamps.
from collections import deque
from dataclasses import dataclass
from datetime import datetime
import time
@dataclass
class Message:
user: str
text: str
timestamp: str = ""
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.now().strftime("%H:%M:%S")
class…Dead Letter Queue Failed Messages List Mock in Python
Implements a simple in-memory dead letter queue to collect, list, and retry failed messages, with JSON serialization for inspection in streaming pipelines.
import json
from collections import deque
class Message:
def __init__(self, message_id, payload, attempts=0):
self.message_id = message_id
self.payload = payload
self.attempts = attempts
def __repr__(self):
return f"Message(id={self.message_id}, attempts={self.attempts})"
c…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.
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…How to Build a Message Stream Queue in Python
A beginner-friendly MessageStream class built on deque that sends messages one at a time, tracks unread counts, and records sent items.
from collections import deque
import time
class MessageStream:
def __init__(self, messages):
self._queue = deque(messages)
self._sent = []
def send_next(self):
if not self._queue:
return None
message = self._queue.popleft()
self._sent.append(message)
…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.
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…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.
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(
…Implement a FIFO Message Queue in Python with deque
This code implements a FIFO (first-in-first-out) message queue class using Python's collections.deque, providing enqueue, dequeue, peek, and size operations.
from collections import deque
class MessageQueue:
def __init__(self):
self.queue = deque()
def enqueue(self, message):
self.queue.append(message)
print(f"Enqueued: {message}")
def dequeue(self):
if self.is_empty():
print("Queue is empty, cannot dequeue.")
…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.
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:
…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.
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…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.
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…Build a queue-based admission control system in Python
Implement a simple bounded-queue admission controller that accepts or rejects incoming requests based on current queue capacity.
from collections import deque
import time
class AdmissionControl:
"""Simple admission control using a bounded queue.
Requests arrive at the queue; they are admitted in FIFO order.
If the queue is full, the incoming request is rejected.
"""
def __init__(self, capacity: int):
self.capacit…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.
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…How to Implement a Rate Limiter in Python
A beginner-friendly Python class that tracks call timestamps with a deque to allow or block calls based on a max rate per time period.
import time
from collections import deque
class RateLimiter:
"""Simple rate limiter for beginners."""
def __init__(self, max_calls: int, period_seconds: float):
self.max_calls = max_calls
self.period = period_seconds
self.calls = deque()
def allow(self) -> bool:
"""Retur…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
- Pick a topic section — strings, lists, files, functions, and more
- Open a sample, read How it works, and copy the code block
- 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.