Reference library

Python Code Samples

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

4 matches
Streaming & messaging easy

How to Mock RabbitMQ Ack Nack Requeue in Python

A mock RabbitMQ channel and consumer that simulates ack, nack, and requeue handling for testing message processing logic without a broker.

rabbitmq testing mock
Python
import json
from collections import deque


class MockChannel:
    def __init__(self):
        self.acked = []
        self.nacked = []
        self.requeued = []

    def basic_ack(self, delivery_tag):
        self.acked.append(delivery_tag)

    def basic_nack(self, delivery_tag, requeue=False):
        self.nacked.…
15 0 Open
Streaming & messaging medium

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.

rabbitmq routing messaging
Python
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…
14 0 Open
Streaming & messaging easy

How to mock RabbitMQ queue binding with routing keys in Python

A mock demonstration of binding a queue to an exchange with multiple routing keys in RabbitMQ using Python and pika, without a real broker connection.

rabbitmq messaging pika
Python
import pika
import sys


def bind_queue_with_routing(channel, queue_name, exchange_name, routing_keys):
    """
    Mock RabbitMQ queue binding with routing keys.
    Prints the binding configuration instead of connecting to a real broker.
    """
    for routing_key in routing_keys:
        binding = {
            "q…
14 0 Open
Streaming & messaging medium

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.

rabbitmq threading qos
Python
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…
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.