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 11, 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:
Newer release available (0.7.1)

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 11, 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.0.tar.gz (31.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.0-py3-none-any.whl (24.5 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: amqpstorm_flask-0.7.0.tar.gz
  • Upload date:
  • Size: 31.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.0.tar.gz
Algorithm Hash digest
SHA256 2621a2671e68d6f0272a2523a06114924bdb73146e50208dde37685acf8ff071
MD5 9e4a755efd3bb3a2d81ce934b6736c91
BLAKE2b-256 afd909fe97017354e22c2cfed995244724c19883df5daa058c9d19bfb54a8c31

See more details on using hashes here.

Provenance

The following attestation bundles were made for amqpstorm_flask-0.7.0.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.0-py3-none-any.whl.

File metadata

  • Download URL: amqpstorm_flask-0.7.0-py3-none-any.whl
  • Upload date:
  • Size: 24.5 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.0-py3-none-any.whl
Algorithm Hash digest
SHA256 a765826fd1a0221d6082b7440c88917d919440a013f08a841ae025a09cce88c4
MD5 562dd35b480436de9fa5afe8ad09ff61
BLAKE2b-256 2909f3081086381f5565ee1fdd0764871d4044f8fc5c5a2d14a0fa173b47be58

See more details on using hashes here.

Provenance

The following attestation bundles were made for amqpstorm_flask-0.7.0-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

0.7.1

2 files

This release

0.7.0 This release

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