AceMQ for Python

A Python client for AceMQ messaging over AMQP, speaking the same wire contract as the Java, Go and .NET libraries: the same reserved headers, the same defaults, the same retry arithmetic. A Python consumer reads what a Java producer writes.

That claim is pinned rather than asserted. The fixtures under tests/fixtures/ were generated by the Java implementation and are committed byte for byte by all five libraries: one file for the headers a message carries, and one for the retry arithmetic, the queue names and the topology. A port that hand-copies either out of documentation acquires a difference nobody notices until two languages disagree in production — which has happened, four times in one afternoon, and is what the conformance suite is for.

pip install --extra-index-url https://acemq.org/pypi/simple/ "acemq-amqp[rabbitmq]==0.7.12"

A message, end to end

import asyncio
from datetime import timedelta

from acemq_amqp import (
    Ack, Message, Topology, accept, connect, exponential_retry, retry,
)


async def main() -> None:
    async with await connect(
        "amqp://guest:guest@localhost:5672/",
        origin="checkout@pod-7",
        retry=exponential_retry(5, timedelta(seconds=1), timedelta(minutes=1)),
    ) as mq:
        await mq.declare(
            Topology()
            .exchange("orders-events", "topic")
            .queue("shipping.orders", dead_letter=True)
            .binding("shipping.orders", "orders-events", "order.placed")
        )

        async def ship(message: Message) -> Ack:
            if not warehouse_is_up():
                return retry(RuntimeError("the warehouse is not answering"))
            place(message.payload)
            return accept()

        consumer = await mq.consume("shipping.orders", ship)

        await mq.publisher("orders-events", "order.placed").send({"id": "7"})
        await asyncio.sleep(5)
        await consumer.close()


asyncio.run(main())

Getting started takes this apart line by line, and the tutorials build it up from nothing in four steps.

What it does for you

An envelope on every message. Identity, type, correlation, causation, attempt count, origin and the time it was first published — the same fields the Java, Go, Ruby and .NET libraries write, so a message crossing between them keeps its history. See the envelope.

Retries that stop. A handler returns retry(), and a policy decides whether there is another attempt and how long to wait. A retry is republished rather than requeued, so the attempt count really advances and lives on the message rather than in the memory of the process that has been failing. Long waits are spent in the broker, on a rung queue whose time-to-live is the wait, so a consumer restart cannot shorten them. See retries, redelivery and shutdown.

Security you have to opt out of. The URL scheme decides whether the connection is encrypted, an amqps:// URL verifies the broker against the machine's trust store with no further configuration, and the way out is a function with a long name rather than a keyword argument somebody can paste. See security.

A core with no dependencies at all. pip install acemq-amqp installs the envelope, the codecs, the retry arithmetic and the naming, and nothing else — not even a UUID library. The broker client is the [rabbitmq] extra and the metrics client is [prometheus], because a service that only reads an AceMQ envelope somebody else delivered should not be made to install an AMQP client to do it.

A blocking API for programs that are not asyncio. acemq_amqp.sync is a facade over the same connection: one loop on a thread of its own, handlers on worker threads, and exactly one implementation of the retry rules underneath. See getting started.

Numbers an operator cannot get from outside. Throughput, failures, handler duration and in-flight count through a three-method Observer, and a health check that asks the broker a real question rather than looking at a socket. See metrics, health and tracing.

Typed throughout. The package ships py.typed, is checked under mypy --strict, and everything on the wire is a frozen dataclass.

The layout

One package, a module per concern:

Everything worth naming is re-exported from the top, so a program imports from one place:

from acemq_amqp import Ack, Envelope, Message, Topology, accept, connect, reject, retry

The exception is the two modules that are a choice rather than the default — acemq_amqp.sync and acemq_amqp.prometheus — and the patterns, which are from acemq_amqp.patterns import ... because a service uses two of the thirteen.

One difference from the other four libraries

Identical, because a message crosses languages: the reserved header names and their types, the defaults applied when they are absent, the retry schedule arithmetic, the {queue}.dlq / {queue}.parked / {queue}.retry.{delay} naming, and the rules for giving up.

Not identical, deliberately: the API shape. Go takes a ctx and returns an error; .NET has IAsyncEnumerable; Python gets async with, frozen dataclasses, keyword arguments and exceptions. Forcing a Java shape onto Python produces a library nobody enjoys using. The contract is portable; the ergonomics are native.

The one place that shows up most is the interceptor. Java has a pair of before-and-after hooks; here an interceptor is a single function handed the message and the rest of the work, because try/finally already runs the way out in the reverse of the way in. See interceptors.

Where to go next

• Tutorials — four of them, in order, each ending with something that runs
• Getting started — the first message, explained
• Publishing and consuming
• Exchanges, queues and bindings
• The envelope — what travels with a message
• Interceptors — the seam for what every message needs
• Security — TLS, trust and credentials
• Retries, redelivery and shutdown
• Codecs — JSON, bytes, text, the five optional formats, and writing your own
• Patterns — the thirteen things every service ends up writing
• Request and reply — a round trip drawn on top of one-way messaging, and what it costs
• Streams — a queue that keeps what it has already handed out
• Metrics, health and tracing
• Testing without a broker
• API reference, generated by pdoc from the docstrings in src/

Status

Pre-1.0 and in development. The API may change before 1.0. The contract layer, the transport and the pattern library are implemented and tested, against a real broker as well as a fake one. The current release is 0.7.12, from https://acemq.org/pypi/simple/ — getting started says how to install from it.