AMQP Service for Flask Apps

A Flask extension for AMQPStorm that provides easy integration with RabbitMQ.

Features

  • Automatic Reconnection: Uses APScheduler to monitor and reconnect to RabbitMQ if the connection drops.
  • Easy Publishing: Simple method to send JSON messages with retry logic.
  • Decorator-based Consumers: Define message consumers using simple decorators.
  • Health Checks: Built-in health check functionality.
  • Prometheus Metrics: Message processing durations per queue, see Metrics.
  • Configuration: Configure via Flask app config or environment variables.

Installation

pip install amqpstorm-flask

Configuration

The extension can be configured via Flask's app.config or environment variables.

Config Key Environment Variable Default Description
MQ_URL MQ_URL None AMQP connection URL (e.g. amqp://guest:guest@localhost:5672/%2f)
MQ_EXCHANGE MQ_EXCHANGE None Default exchange name
AMQP_STORM_APSCHEDULER 1 Use APScheduler for health checks and consumers (recommended)
FILTER_LOGS 1 Filter out noisy connection logs
MQ_MAX_CONSUMER_IDLE_TIME 300 Max idle time in seconds before reconnecting
MQ_DELIMITER . Delimiter used to generate queue names from function names
MQ_QUEUES None Comma-separated list of enabled queues (if set, only these will start)

Usage

Initialization

from flask import Flask
from amqpstorm_flask import RabbitMQ

app = Flask(__name__)
app.config["MQ_URL"] = "amqp://guest:guest@localhost:5672/%2f"
app.config["MQ_EXCHANGE"] = "my_exchange"

mq = RabbitMQ(app)

Alternatively, use the factory pattern:

mq = RabbitMQ()
# ... later ...
mq.init_app(app)

Standalone Usage (Without Flask)

If you are not using Flask or want to initialize it manually without the init_app helper:

from amqpstorm_flask import RabbitMQ

mq = RabbitMQ(
    mq_url="amqp://guest:guest@localhost:5672/%2f",
    mq_exchange="my_exchange"
)
# Manual start is required if init_app is not used
mq.start()

Sending Messages

mq.send(
    body={"hello": "world"},
    routing_key="events.user.created",
    exchange_type="topic"
)

Consuming Messages

By default, the decorated function receives routing_key, body, and message_id.

@mq.queue(routing_key="events.user.#")
def handle_user_events(routing_key, body, message_id):
    print(f"Received event {routing_key}: {body}")

To receive the full AMQPStorm message object:

@mq.queue(routing_key="events.user.#", full_message_object=True)
def handle_user_events(message):
    print(f"Received body: {message.body}")
    # Manual acknowledgment if auto_ack is False
    # message.ack()

Health Check

You can use the check_health method to implement a health check endpoint:

@app.route("/health")
def health_check():
    is_ok, message = mq.check_health()
    if is_ok:
        return {"status": "ok"}, 200
    return {"status": "error", "message": message}, 503

Advanced Queue Configuration

The @mq.queue decorator supports several parameters:

  • routing_key: String or list of strings.
  • queue_name: Custom queue name (defaults to function name with _ replaced by MQ_DELIMITER).
  • exchange_type: Default is "topic".
  • auto_ack: Whether to automatically acknowledge messages.
  • prefetch_count: Number of messages to prefetch (default 1).
  • queue_arguments: Dictionary of arguments for queue declaration (default {"x-queue-type": "quorum"}).
  • full_message_object: If True, the decorated function receives the message object instead of unpacked values.

Metrics

Every consumer is instrumented with three prometheus histograms, labeled with env, service_name and queue:

Metric Measures
message_processing_active_duration_seconds Time spent inside the consumer callback
message_processing_waiting_duration_seconds Publication until the start of processing
message_processing_total_duration_seconds Publication until the message is acknowledged

The three are independent observations. total is only observed once the message reaches a terminal state: an ack, or a nack/reject with requeue=False. A nack that requeues produces no total, because the message will be delivered again, and the eventual ack of that delivery reports the total from its own publication time. Consumers using auto_ack never settle the message themselves, so for them the end of the callback counts as terminal.

waiting and total need the publication time. send() stamps it as an x-published-at header (epoch seconds, float); messages without it fall back to the AMQP timestamp property, which only has one second resolution. Messages with neither get no waiting or total observation.

Environment Variable Default Description
AMQP_METRICS_ENABLED 1 Set to 0 to skip consumer instrumentation entirely
MESSAGE_PROCESSING_ACTIVE_DURATION_BUCKETS .005 … 10 Comma-separated bucket bounds
MESSAGE_PROCESSING_WAITING_DURATION_BUCKETS .01 … 600 Comma-separated bucket bounds
MESSAGE_PROCESSING_TOTAL_DURATION_BUCKETS .01 … 600 Comma-separated bucket bounds

NOMAD_JOB_NAME and NOMAD_GROUP_NAME provide the env and service_name label values.

Project links

Data verified by PyPI on Aug 13, 2026
Data provided by the project maintainers, verified at the time the release was uploaded to PyPI.

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for inuits from gravatar.com inuits

Credits

Author: Inuits

GitHub Statistics

Data verified by PyPI on Aug 13, 2026
The GitHub source repository was provided by the project maintainers and verified by PyPI at the time of upload. Stars, forks, and open issues/PRs are derived from that repository and have not been independently verified.

License

GNU General Public License v2 (GPLv2) (GNU GENERAL PUBLIC LICENSE)

Tags

amqpstorm_flask amqpstorm-flask

Classifiers

Report project as malware

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

amqpstorm_flask-0.7.1.tar.gz (32.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

amqpstorm_flask-0.7.1-py3-none-any.whl (24.9 kB view details)

Uploaded Python 3

File details

Details for the file amqpstorm_flask-0.7.1.tar.gz.

File metadata

  • Download URL: amqpstorm_flask-0.7.1.tar.gz
  • Upload date:
  • Size: 32.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for amqpstorm_flask-0.7.1.tar.gz
Algorithm Hash digest
SHA256 d0d1fdedfd8da090b976835a9cd0e20149fea05dcd16d79e550a307f8b17e2fa
MD5 20175e68408d5f8ce023430124cf87c3
BLAKE2b-256 85af2183b7c0be4387c659d9a35f3e3246570b6b0725de5dd6ae44cb7a3a8674

See more details on using hashes here.

Provenance

The following attestation bundles were made for amqpstorm_flask-0.7.1.tar.gz:

Publisher: python-publish.yml on inuits/amqpstorm-flask

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file amqpstorm_flask-0.7.1-py3-none-any.whl.

File metadata

  • Download URL: amqpstorm_flask-0.7.1-py3-none-any.whl
  • Upload date:
  • Size: 24.9 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for amqpstorm_flask-0.7.1-py3-none-any.whl
Algorithm Hash digest
SHA256 ed87cf2b29c31d7a5a257efa1625db0b313bd004107f785efa1907321a46e02c
MD5 fd8307d0be648f1fa929f3964240df4c
BLAKE2b-256 33a9eb6cf969cc5c115e777b8633b4666efa933ce11ee3bed33ba5d8e20efaae

See more details on using hashes here.

Provenance

The following attestation bundles were made for amqpstorm_flask-0.7.1-py3-none-any.whl:

Publisher: python-publish.yml on inuits/amqpstorm-flask

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Release history Release notifications | RSS feed

This release

0.7.1 This release

2 files

0.7.0

2 files

0.6.2

2 files

0.6.1

2 files

Pre-release

0.6.0.dev0

2 files

0.5.4

2 files

0.5.3

2 files

0.5.2

2 files

0.5.1

2 files

0.4.8

2 files

0.4.7

2 files

0.4.6

2 files

0.4.4

2 files

0.4.3

2 files

0.4.2

2 files

0.4.1

2 files

0.4.0

2 files

0.3.4

2 files

0.3.2

2 files

0.3.1

2 files

0.2.3

2 files

0.2.2

2 files

0.2.1

2 files

0.2.0

2 files

0.1.9

2 files

0.1.8

2 files

0.1.7

2 files

0.1.6

2 files

0.1.5

2 files

0.1.4

2 files

0.1.3

2 files

0.1.2

2 files

0.1.1

2 files

0.1.0

2 files

0.0.4

2 files

0.0.3

2 files

0.0.2

2 files

0.0.1

2 files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page