Flask-RMQ 0.1.0

Production-oriented RabbitMQ integration for Flask, built on Pika

---

---

Documentation: https://flask-rmq.rdd-lab.com/

Source Code: https://github.com/RDDLab/Flask-RMQ

---

Flask-RMQ

Flask-RMQ provides typed RabbitMQ wrappers and tools for Flask applications using Pika.

It is not a task queue or a Celery replacement. It is a lightweight integration layer for services that need explicit message contracts, reliable publication, long-running consumers, topology management, and multiple independent RabbitMQ brokers.

Main features:

• standard Flask extension and app-factory integration;
• Click commands under the native flask rmq CLI group;
• separate thread-local producer and consumer connections;
• publisher confirms, persistent delivery, and mandatory=True routing;
• bounded producer retry and consumer exponential backoff;
• durable queues and dead-letter routing through QueueConfig;
• per-alias topology and consumer registries;
• optional Prometheus instrumentation;
• Python typing with a py.typed marker.

Installation

Install Flask-RMQ with pip or your preferred Python package manager:

pip install Flask-RMQ

Prometheus support is optional:

pip install 'Flask-RMQ[metrics]'

Flask-RMQ supports Python 3.10–3.14 and Flask 3.0–3.1.

Minimal application

import json

from flask import Flask
from flask_rmq import (
    Consumer,
    FlaskRMQ,
    Producer,
    get_consumers_registry,
    get_setup_registry,
)

rmq = FlaskRMQ()
producer = Producer(queue='events')
consumer = Consumer(queue='events')


@consumer
def handle_event(channel, method, properties, body: bytes) -> None:
    print(json.loads(body))
    channel.basic_ack(delivery_tag=method.delivery_tag)


def setup_topology(channel) -> None:
    channel.queue_declare(queue='events', durable=True)


def create_app() -> Flask:
    app = Flask(__name__)
    app.config['RABBITMQ_CONNECTIONS'] = {
        'default': {'HOST': 'localhost'},
    }
    rmq.init_app(app)
    with app.app_context():
        get_setup_registry().register(setup_topology)
        get_consumers_registry().register(consumer)
    return app

flask --app your_app:create_app rmq check
flask --app your_app:create_app rmq setup
flask --app your_app:create_app rmq consume

Publish from a Flask route or service:

producer.publish(json.dumps({'event': 'created'}))

The complete runnable producer/consumer project is available in examples/basic_app.

Tests

python -m venv .venv
. .venv/bin/activate
pip install -e '.[dev,metrics]'
pytest
ruff check .
python -m build

The unit test suite does not require a live RabbitMQ broker. Use the example project for an end-to-end broker test.

Documentation portal

The VuePress portal contains complete English and Russian guides for configuration, producers, consumers, topology, CLI, reliability, metrics, the example project, and migration from Django-RMQ.

npm install
npm run docs:dev
npm run docs:build

MIT licensed.