Python Code
Samples
Medium snippets you can copy, study, and run in the browser editor.
How to Mock AWS SQS Send Receive Delete in Python
Build an in-memory mock of the SQS send, receive, and delete message flow for local testing.
import json
from collections import deque
from uuid import uuid4
class MockSQSQueue:
def __init__(self, name):
self.name = name
self._messages = deque()
self._in_flight = {}
def send_message(self, body, attributes=None):
message_id = str(uuid4())
message = {
…How to Build a Producer-Consumer Pattern with asyncio.Queue in Python
This code implements a classic producer-consumer pattern using asyncio.Queue to coordinate one producer task that generates items and two consumer tasks that process them concurrently, with a sentinel value to signal completion.
import asyncio
import random
async def producer(queue, item_count):
for i in range(item_count):
item = random.randint(1, 100)
await queue.put(item)
print(f"Produced: {item}")
await asyncio.sleep(0.1)
await queue.put(None) # Sentinel to signal end
async def consumer(queue, n…How to Implement a Batch Requests Flush Interval in Python
A simple async batcher that accumulates items and flushes them either when a max batch size is reached or after a time-based flush interval.
import asyncio
from collections import deque
class Batcher:
def __init__(self, flush_interval=0.5, max_batch=5):
self.flush_interval = flush_interval
self.max_batch = max_batch
self.queue = deque()
self.lock = asyncio.Lock()
async def add(self, item):
async with self.l…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.
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}")
…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.
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…How to Implement At-Least-Once Delivery with Acknowledgment in Python
This code demonstrates a mock message broker with at-least-once delivery, including retry logic and acknowledgment after successful processing.
import time
import uuid
from collections import deque
class MockMessageBroker:
def __init__(self):
self.queue = deque()
self.acked = set()
def publish(self, payload: str) -> str:
msg_id = str(uuid.uuid4())
self.queue.append((msg_id, payload))
return msg_id
def po…How to Implement Backpressure Pause Producer with a Bounded Queue in Python
Places a Producer thread that sends items into a bounded queue with backpressure: on Full, it pauses to let the consumer catch up.
import threading
import time
import queue
import random
class Producer:
def __init__(self, q):
self.q = q
self.running = True
def produce(self):
while self.running:
item = random.randint(1, 100)
try:
self.q.put(item, timeout=0.5)
…How to Simulate RabbitMQ Exchange Routing in Python
Simulate RabbitMQ exchange routing using a nested dict, matching routing keys against patterns like error.* and info.# to return bound queues.
from collections import defaultdict
def route_message(exchanges, exchange_name, routing_key):
"""
Simulate RabbitMQ exchange routing using a nested dict structure.
Returns list of queue names that match the routing key.
"""
queues = exchanges.get(exchange_name, {})
matched = []
for pa…Implement a retry queue with visibility timeout in Python
This code simulates a message queue with a visibility timeout, allowing messages to be retried if not deleted before the timeout expires.
import time
from collections import deque
class SimpleQueue:
def __init__(self, visibility_timeout=2):
self.queue = deque()
self.in_flight = {}
self.visibility_timeout = visibility_timeout
def send(self, message):
self.queue.append(message)
def receive(self):
if …Simulate RabbitMQ QoS Prefetch Count in Python
Mocks RabbitMQ QoS prefetch semantics using threading and a queue to cap concurrent unacked message processing per worker.
import threading
import time
import queue
class RabbitMQMock:
def __init__(self, prefetch_count=1):
self.prefetch_count = prefetch_count
self.channel_queue = queue.Queue()
self.currently_processing = 0
self.lock = threading.Lock()
def start_consuming(self, messages, worker_co…How to Implement Probabilistic Early Expiration in Python
A Python mock of probabilistic early expiration for caches, using a heap-based expiry queue and random eviction to approximate cache stampede protection.
import heapq
import random
import time
class ProbabilisticEarlyExpirationMock:
def __init__(self, capacity=1024, expiration_probability=0.1):
self.capacity = capacity
self.expiration_probability = expiration_probability
self._items = {}
self._expiry_heap = []
self._next_id…How to Mock Redis Pub/Sub in Python
Test Redis pub/sub logic without a live server using an in-memory fake that queues published messages per channel.
import redis
import time
import threading
class MockRedisPubSub:
def __init__(self):
self.channels = {}
def publish(self, channel, message):
if channel not in self.channels:
return 0
for subscriber in self.channels[channel]:
subscriber.put(message)
ret…How to Mock a Redis Transaction with MULTI/EXEC in Python
A minimal in-memory mock of Redis MULTI/EXEC transactions that queues commands and applies them atomically on EXEC.
class RedisTransactionMock:
def __init__(self):
self.data = {}
self.queue = []
self.in_transaction = False
def multi(self):
self.in_transaction = True
self.queue = []
return "OK"
def set(self, key, value):
if self.in_transaction:
self.qu…How to implement a write-behind cache with async queue in Python
Build an async write-behind cache that queues writes in memory and flushes them in batches to persistent storage.
import asyncio
from collections import deque
from dataclasses import dataclass
@dataclass
class CacheEntry:
key: str
value: str
class WriteBehindCache:
def __init__(self, flush_interval=1.0):
self.cache = {}
self.queue = deque()
self.flush_interval = flush_interval
self._f…How to Send Messages to a Dead Letter Queue in Python
Simulates a poison message queue that retries failed messages up to a limit before moving them to a dead letter queue.
import json
class PoisonMessageQueue:
def __init__(self, max_retries=3):
self.dlq = []
self.max_retries = max_retries
self.processed_count = 0
self.failed_count = 0
def process_message(self, message_body):
if "poison" in message_body:
self.failed_count += 1…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.