Metadata-Version: 2.5
Name: acemq-amqp
Version: 0.7.6
Summary: AceMQ messaging over AMQP for Python, speaking the same wire contract as the Java, Go and .NET libraries
Project-URL: Homepage, https://acemq.org
Project-URL: Source, https://github.com/AceMQ-Company/acemq-python-amqp
Project-URL: Issues, https://github.com/AceMQ-Company/acemq-python-amqp/issues
Author: AceMQ
License:                                  Apache License
                                   Version 2.0, January 2004
                                http://www.apache.org/licenses/
        
           TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
        
           1. Definitions.
        
              "License" shall mean the terms and conditions for use, reproduction,
              and distribution as defined by Sections 1 through 9 of this document.
        
              "Licensor" shall mean the copyright owner or entity authorized by
              the copyright owner that is granting the License.
        
              "Legal Entity" shall mean the union of the acting entity and all
              other entities that control, are controlled by, or are under common
              control with that entity. For the purposes of this definition,
              "control" means (i) the power, direct or indirect, to cause the
              direction or management of such entity, whether by contract or
              otherwise, or (ii) ownership of fifty percent (50%) or more of the
              outstanding shares, or (iii) beneficial ownership of such entity.
        
              "You" (or "Your") shall mean an individual or Legal Entity
              exercising permissions granted by this License.
        
              "Source" form shall mean the preferred form for making modifications,
              including but not limited to software source code, documentation
              source, and configuration files.
        
              "Object" form shall mean any form resulting from mechanical
              transformation or translation of a Source form, including but
              not limited to compiled object code, generated documentation,
              and conversions to other media types.
        
              "Work" shall mean the work of authorship, whether in Source or
              Object form, made available under the License, as indicated by a
              copyright notice that is included in or attached to the work
              (an example is provided in the Appendix below).
        
              "Derivative Works" shall mean any work, whether in Source or Object
              form, that is based on (or derived from) the Work and for which the
              editorial revisions, annotations, elaborations, or other modifications
              represent, as a whole, an original work of authorship. For the purposes
              of this License, Derivative Works shall not include works that remain
              separable from, or merely link (or bind by name) to the interfaces of,
              the Work and Derivative Works thereof.
        
              "Contribution" shall mean any work of authorship, including
              the original version of the Work and any modifications or additions
              to that Work or Derivative Works thereof, that is intentionally
              submitted to Licensor for inclusion in the Work by the copyright owner
              or by an individual or Legal Entity authorized to submit on behalf of
              the copyright owner. For the purposes of this definition, "submitted"
              means any form of electronic, verbal, or written communication sent
              to the Licensor or its representatives, including but not limited to
              communication on electronic mailing lists, source code control systems,
              and issue tracking systems that are managed by, or on behalf of, the
              Licensor for the purpose of discussing and improving the Work, but
              excluding communication that is conspicuously marked or otherwise
              designated in writing by the copyright owner as "Not a Contribution."
        
              "Contributor" shall mean Licensor and any individual or Legal Entity
              on behalf of whom a Contribution has been received by Licensor and
              subsequently incorporated within the Work.
        
           2. Grant of Copyright License. Subject to the terms and conditions of
              this License, each Contributor hereby grants to You a perpetual,
              worldwide, non-exclusive, no-charge, royalty-free, irrevocable
              copyright license to reproduce, prepare Derivative Works of,
              publicly display, publicly perform, sublicense, and distribute the
              Work and such Derivative Works in Source or Object form.
        
           3. Grant of Patent License. Subject to the terms and conditions of
              this License, each Contributor hereby grants to You a perpetual,
              worldwide, non-exclusive, no-charge, royalty-free, irrevocable
              (except as stated in this section) patent license to make, have made,
              use, offer to sell, sell, import, and otherwise transfer the Work,
              where such license applies only to those patent claims licensable
              by such Contributor that are necessarily infringed by their
              Contribution(s) alone or by combination of their Contribution(s)
              with the Work to which such Contribution(s) was submitted. If You
              institute patent litigation against any entity (including a
              cross-claim or counterclaim in a lawsuit) alleging that the Work
              or a Contribution incorporated within the Work constitutes direct
              or contributory patent infringement, then any patent licenses
              granted to You under this License for that Work shall terminate
              as of the date such litigation is filed.
        
           4. Redistribution. You may reproduce and distribute copies of the
              Work or Derivative Works thereof in any medium, with or without
              modifications, and in Source or Object form, provided that You
              meet the following conditions:
        
              (a) You must give any other recipients of the Work or
                  Derivative Works a copy of this License; and
        
              (b) You must cause any modified files to carry prominent notices
                  stating that You changed the files; and
        
              (c) You must retain, in the Source form of any Derivative Works
                  that You distribute, all copyright, patent, trademark, and
                  attribution notices from the Source form of the Work,
                  excluding those notices that do not pertain to any part of
                  the Derivative Works; and
        
              (d) If the Work includes a "NOTICE" text file as part of its
                  distribution, then any Derivative Works that You distribute must
                  include a readable copy of the attribution notices contained
                  within such NOTICE file, excluding those notices that do not
                  pertain to any part of the Derivative Works, in at least one
                  of the following places: within a NOTICE text file distributed
                  as part of the Derivative Works; within the Source form or
                  documentation, if provided along with the Derivative Works; or,
                  within a display generated by the Derivative Works, if and
                  wherever such third-party notices normally appear. The contents
                  of the NOTICE file are for informational purposes only and
                  do not modify the License. You may add Your own attribution
                  notices within Derivative Works that You distribute, alongside
                  or as an addendum to the NOTICE text from the Work, provided
                  that such additional attribution notices cannot be construed
                  as modifying the License.
        
              You may add Your own copyright statement to Your modifications and
              may provide additional or different license terms and conditions
              for use, reproduction, or distribution of Your modifications, or
              for any such Derivative Works as a whole, provided Your use,
              reproduction, and distribution of the Work otherwise complies with
              the conditions stated in this License.
        
           5. Submission of Contributions. Unless You explicitly state otherwise,
              any Contribution intentionally submitted for inclusion in the Work
              by You to the Licensor shall be under the terms and conditions of
              this License, without any additional terms or conditions.
              Notwithstanding the above, nothing herein shall supersede or modify
              the terms of any separate license agreement you may have executed
              with Licensor regarding such Contributions.
        
           6. Trademarks. This License does not grant permission to use the trade
              names, trademarks, service marks, or product names of the Licensor,
              except as required for reasonable and customary use in describing the
              origin of the Work and reproducing the content of the NOTICE file.
        
           7. Disclaimer of Warranty. Unless required by applicable law or
              agreed to in writing, Licensor provides the Work (and each
              Contributor provides its Contributions) on an "AS IS" BASIS,
              WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
              implied, including, without limitation, any warranties or conditions
              of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
              PARTICULAR PURPOSE. You are solely responsible for determining the
              appropriateness of using or redistributing the Work and assume any
              risks associated with Your exercise of permissions under this License.
        
           8. Limitation of Liability. In no event and under no legal theory,
              whether in tort (including negligence), contract, or otherwise,
              unless required by applicable law (such as deliberate and grossly
              negligent acts) or agreed to in writing, shall any Contributor be
              liable to You for damages, including any direct, indirect, special,
              incidental, or consequential damages of any character arising as a
              result of this License or out of the use or inability to use the
              Work (including but not limited to damages for loss of goodwill,
              work stoppage, computer failure or malfunction, or any and all
              other commercial damages or losses), even if such Contributor
              has been advised of the possibility of such damages.
        
           9. Accepting Warranty or Additional Liability. While redistributing
              the Work or Derivative Works thereof, You may choose to offer,
              and charge a fee for, acceptance of support, warranty, indemnity,
              or other liability obligations and/or rights consistent with this
              License. However, in accepting such obligations, You may act only
              on Your own behalf and on Your sole responsibility, not on behalf
              of any other Contributor, and only if You agree to indemnify,
              defend, and hold each Contributor harmless for any liability
              incurred by, or claims asserted against, such Contributor by reason
              of your accepting any such warranty or additional liability.
        
           END OF TERMS AND CONDITIONS
        
           APPENDIX: How to apply the Apache License to your work.
        
              To apply the Apache License to your work, attach the following
              boilerplate notice, with the fields enclosed by brackets "[]"
              replaced with your own identifying information. (Don't include
              the brackets!)  The text should be enclosed in the appropriate
              comment syntax for the file format. We also recommend that a
              file or class name and description of purpose be included on the
              same "printed page" as the copyright notice for easier
              identification within third-party archives.
        
           Copyright [yyyy] [name of copyright owner]
        
           Licensed under the Apache License, Version 2.0 (the "License");
           you may not use this file except in compliance with the License.
           You may obtain a copy of the License at
        
               http://www.apache.org/licenses/LICENSE-2.0
        
           Unless required by applicable law or agreed to in writing, software
           distributed under the License is distributed on an "AS IS" BASIS,
           WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
           See the License for the specific language governing permissions and
           limitations under the License.
License-File: LICENSE
Keywords: acemq,amqp,messaging,rabbitmq
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: System :: Networking
Classifier: Typing :: Typed
Requires-Python: >=3.10
Provides-Extra: avro
Requires-Dist: fastavro>=1.9; extra == 'avro'
Provides-Extra: crypto
Requires-Dist: cryptography>=42.0; extra == 'crypto'
Provides-Extra: dev
Requires-Dist: aio-pika>=9.4; extra == 'dev'
Requires-Dist: cryptography>=42.0; extra == 'dev'
Requires-Dist: fastavro>=1.9; extra == 'dev'
Requires-Dist: mypy>=1.11; extra == 'dev'
Requires-Dist: opentelemetry-api>=1.20; extra == 'dev'
Requires-Dist: opentelemetry-sdk>=1.20; extra == 'dev'
Requires-Dist: pdoc==15.0.1; extra == 'dev'
Requires-Dist: prometheus-client>=0.20; extra == 'dev'
Requires-Dist: protobuf>=4.25; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: pyyaml>=6.0; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Requires-Dist: tomli-w>=1.0; extra == 'dev'
Requires-Dist: tomli>=2.0; (python_version < '3.11') and extra == 'dev'
Requires-Dist: types-protobuf; extra == 'dev'
Requires-Dist: types-pyyaml; extra == 'dev'
Provides-Extra: opentelemetry
Requires-Dist: opentelemetry-api>=1.20; extra == 'opentelemetry'
Provides-Extra: postgres
Requires-Dist: psycopg[binary]>=3.1; extra == 'postgres'
Provides-Extra: prometheus
Requires-Dist: prometheus-client>=0.20; extra == 'prometheus'
Provides-Extra: protobuf
Requires-Dist: protobuf>=4.25; extra == 'protobuf'
Provides-Extra: rabbitmq
Requires-Dist: aio-pika>=9.4; extra == 'rabbitmq'
Provides-Extra: toml
Requires-Dist: tomli-w>=1.0; extra == 'toml'
Requires-Dist: tomli>=2.0; (python_version < '3.11') and extra == 'toml'
Provides-Extra: yaml
Requires-Dist: pyyaml>=6.0; extra == 'yaml'
Description-Content-Type: text/markdown

# AceMQ for Python

[![license](https://img.shields.io/badge/license-Apache--2.0-green)](LICENSE)
[![Python](https://img.shields.io/badge/Python-3.10%2B-3776AB)](#requirements)
[![brokers](https://img.shields.io/badge/broker-RabbitMQ-lightgrey)](#requirements)

A Python client for AceMQ messaging over AMQP, speaking the same wire contract
as the [Java](https://github.com/AceMQ-Company/acemq-java-amqp),
[Go](https://github.com/AceMQ-Company/acemq-go-amqp) and
[.NET](https://github.com/AceMQ-Company/acemq-dotnet-amqp) libraries: the same
reserved headers, the same defaults, the same retry arithmetic. A Python
consumer reads what a Java producer writes, and the fixtures generated from the
Java implementation pin that rather than leaving it to be discovered in
production.

> **Status: in build.** The contract layer, the transport and the pattern
> library are implemented and tested, against a real broker as well as a fake
> one. Nothing is published to PyPI yet.

## Sending and receiving

```python
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:
        # Everything this service needs the broker to have, in one description
        # somebody can read before it is applied.
        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())
```

The transport is an extra, because reading an AceMQ envelope should not require
installing a broker client:

```bash
pip install acemq-amqp                # the contract, no dependencies at all
pip install "acemq-amqp[rabbitmq]"    # and a way to talk to a broker
pip install "acemq-amqp[prometheus]"  # and a registry to report into
pip install "acemq-amqp[yaml]"        # and one of the five optional formats
```

The formats go the same way, one extra each — `[yaml]`, `[toml]`, `[protobuf]`,
`[avro]` — because a service that speaks YAML has no reason to install a
protobuf runtime. XML has no extra: it is written against the standard library.

### Publishing a batch

```python
results = await mq.publisher("orders-events", "order.placed").send_all(orders)
```

Every message goes out before any confirm is awaited, and only then are all of
them checked together. A loop calling `send` waits for the broker between one
message and the next; this does not, and the results still come back in the
order the payloads did. It is not atomic — AMQP has no such thing — so a partial
failure raises `PublishError` naming how many messages were not confirmed and
how many were, because a caller told only "it failed" resends messages that
already arrived. How many are unconfirmed at once is bounded by the connection's
`max_outstanding_publishes`, a thousand by default and the same number Java and
.NET default to, so a wide batch becomes backpressure rather than memory growth.
See [publishing](docs/publishing.md#several-at-once) and
[back pressure](docs/publishing.md#back-pressure).

### Not running an event loop?

```python
from acemq_amqp import Topology, accept, sync

with sync.connect("amqp://guest:guest@localhost:5672/") as mq:
    mq.declare(Topology().queue("shipping.orders", dead_letter=True))
    mq.publisher(routing_key="shipping.orders").send({"id": "7"})

    with mq.consume("shipping.orders", lambda message: accept()):
        ...
```

It is a facade, not a second implementation: a loop runs on a thread of its own
and every call is handed to it, so the envelope rules and the retry engine are
the ones above rather than a copy that can drift. Handlers run on a worker
thread, so a blocking handler blocks nothing but itself.

## Reaching a real broker: TLS and credentials

**The URL scheme decides whether the connection is encrypted.** `amqps://` is,
`amqp://` is not, because that is the part somebody reads in a configuration
file and believes. An `amqps://` URL needs nothing else: the broker is verified
against the machine's trust store and nothing older than TLS 1.2 is spoken.

```python
mq = await connect("amqps://broker.internal:5671/")
```

Most brokers are not on a public certificate authority, because no public
authority issues certificates for `broker.internal`. Name the authority instead,
and keep the password out of the URL while you are there:

```python
from acemq_amqp import Security, connect, credentials_from_environment

mq = await connect(
    "amqps://broker.internal:5671/",
    security=Security(
        certificate_authority="/etc/acemq/ca.crt",
        credentials=credentials_from_environment("MQ_USER", "MQ_PASSWORD"),
    ),
)
```

Naming an authority **replaces** the system trust store rather than adding to
it. That is the point: a broker holding a certificate from a public authority is
not your broker, and the several hundred authorities a machine trusts by default
are several hundred ways to be wrong. A self-signed broker certificate works
here too — put the broker's own certificate in the file and it becomes the one
authority you trust.

| | |
|---|---|
| `certificate_authority` | The PEM authority to verify the broker against, and no other |
| `client_certificate` / `client_key` | What to present, for a broker that authenticates clients by certificate |
| `client_key_password` | The passphrase on that key. Kept out of `repr` |
| `server_name` | The name to check the certificate against, when it is not the one in the URL — an IP address, a tunnel, a container's internal name |
| `credentials` | The login, or a source asked for it at connection time |

Everything above works the same on the blocking API — `sync.connect(url,
security=...)` — because a program that is not running an event loop has exactly
the same broker to reach.

### Credentials that do not end up in a log

A password in a connection string reaches every log line, crash report and `ps`
listing that URL ever appears in. Supply it separately and the URL carries a host
and nothing else:

```python
from acemq_amqp import Credentials, credentials_from_environment, credentials_from_file

Credentials("app", password)                        # from wherever you already had it
credentials_from_environment("MQ_USER", "MQ_PASSWORD")
credentials_from_file("/run/secrets/broker", username="app")
```

`Credentials` never renders its secret — `repr`, `str` and an f-string all give
back `Credentials(username='app', secret=<secret>)` — and a test in the suite
asserts it, because the whole value of the type is that it holds under a
`print()` somebody added at three in the morning.

The two file-reading sources are read **each time a connection is made**, not
once at import. That is what makes a password rotated by a sidecar or a
remounted Kubernetes secret take effect without a restart. A source is just a
callable, so `lambda: Credentials("app", vault.read())` is a complete
implementation of one.

### The way out, and what it costs

```python
from acemq_amqp import without_verifying_the_broker

mq = await connect("amqps://localhost:5671/", security=without_verifying_the_broker())
```

This encrypts the traffic and checks nothing at all about who is on the other
end. Any certificate is accepted, from any issuer, for any name, so somebody who
can answer on the address in your URL receives every message you publish and
every password you log in with — over a connection that looks encrypted in every
log and every metric.

It is a function with a long name rather than a `verify=False` on purpose: there
is no keyword argument anywhere in this library that turns verification off, so
it should be impossible to end up here by pasting one, and hard to leave in a
file nobody rereads. **It is wrong in production, always.** The alternative is
one line — `Security(certificate_authority="ca.crt")` — and gives back
everything this gives up.

Settings that describe TLS are **refused against an `amqp://` URL** rather than
ignored, and that is deliberate too: a service that was handed a certificate
authority, connected in plaintext and reported success is the failure this whole
module exists to prevent. Credentials alone are welcome on either scheme.

### Certificates for a broker on a laptop

```bash
pip install "acemq-amqp[crypto]"
python -m acemq_amqp.devcerts --directory certs --broker localhost
```

An authority, a broker certificate, a client certificate and a `rabbitmq.conf`
pointing the broker at them — the same file names Go's `acemq-certs` writes, so
it is a drop-in replacement for it.

Everything it writes carries `ACEMQ DEVELOPMENT ONLY - DO NOT TRUST` in its
subject organisation, and **this library refuses any certificate carrying it,
however trust is configured — `without_verifying_the_broker()` included**. That
is not a warning in a docstring; it is the mechanism. A generated authority's
private key sits next to its certificate and usually ends up in a repository, so
a development certificate that could reach production would be an authority
anybody who can read that repository can issue against, and the connection would
succeed. Java, Go, Ruby and .NET stamp the same string and enforce it the same
way.

The way through, for the one place it belongs:

```python
security=Security(certificate_authority="certs/ca.crt", allow_development_certificates=True)
```

A named argument a reviewer will see, and one more thing to `grep` for in a
deployed configuration. See [Security](docs/security.md#development-certificates).

## What is identical, and what is not

**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, the queue type and arguments each of those is declared with, the rules
for giving up, and where a request says to send its answer — the `acemq-reply-to`
header and AMQP's own `reply-to` property, both written, header read first.

**Not identical**, deliberately: the API shape. Go gets `ctx`, .NET gets
`IAsyncEnumerable`, and Python gets dataclasses, `async with` and type hints.
Forcing a Java shape onto Python produces a library nobody enjoys using. The
contract is portable; the ergonomics are native.

### The envelope

| Header | |
|---|---|
| `x-acemq-id` | The message identifier, and the default idempotency key |
| `x-acemq-type` | The logical type, falling back to the routing key |
| `x-acemq-version` | Schema version, from 1 |
| `x-acemq-correlation` | Defaults to the id, so a chain has something to copy |
| `x-acemq-causation` | The message that caused this one |
| `x-acemq-attempt` | Delivery attempt, from 1 |
| `x-acemq-first-seen` | Epoch **milliseconds** of the first publish |
| `x-acemq-origin` | `service@host` |
| `x-acemq-error` | Why it was dead-lettered |
| `x-acemq-claim` | Where the payload is, when it is stored outside the message |
| `x-acemq-route` / `-position` / `-id` | A route declared in advance, the step it is at, and the run — Java's `Pipeline` form of a routing slip |

Application headers are kept apart from these. A reserved name in your own
headers is refused rather than dropped — silently discarding a header somebody
set is worse than saying no — and unknown `x-acemq-` names from a newer version
of another language's library are not handed back as yours.

### What a handler can answer

```python
from acemq_amqp import accept, park, reject, retry

accept()                 # done. Acknowledged and gone
retry(error)             # try again, if the policy has an attempt left
reject(error)            # do not try again. Straight to {queue}.dlq
park(error)              # unreadable. Straight to {queue}.parked
```

Four functions returning an `Ack`, and the error travels with the value into
`x-acemq-error`, so whoever drains the dead letters reads *why* rather than only
*that*. `reject` is *understood, and refused*; `park` is *never going to be
understood* — which is the same conclusion the engine reaches on its own for a
body no codec will decode, and both land in the same queue for the same reason.
An exception escaping the handler is read as `retry`, because in Python an
exception is how a thing says it failed. See
[consuming](docs/consuming.md#the-decision-is-the-return-value).

### Retry, and where a message goes when it runs out

```python
policy = exponential_retry(5, timedelta(seconds=1), timedelta(minutes=1))
policy = policy.give_up_after(timedelta(hours=6))
```

`schedule()` shows the delays without jitter, which is what to read when
deciding whether a policy is the one you meant. Jitter moves a delay **both
ways**: one-sided jitter only ever delays, which turns a thundering herd into a
slower thundering herd.

Giving up on **age** as well as attempts is the honest limit when a queue has
been paused — a message can be on attempt one and four days old.

A handler that returns `retry()` gets another delivery if the policy has one
left. When it does not, the message is republished to `{queue}.dlq` with
`x-acemq-error` saying which limit it hit and what the last failure was —
`exhausted 5 attempts: RuntimeError: the warehouse is not answering` — and the
original is acknowledged, because it has already been safely put somewhere else.
Raising `FatalError`, or returning `retry(FatalError(...))`, skips the attempts
that are left: they would all fail the same way.

A retry is republished rather than requeued, so `x-acemq-attempt` really
advances and the count lives on the message rather than in the memory of the
process that has been failing. The trade is that a retried message goes to the
back of its queue rather than the front.

### Where the waiting happens

A short wait is spent in the consumer, which holds the delivery and one prefetch
slot. A long one is spent in the broker, on a `{queue}.retry.{delay}` queue whose
`x-message-ttl` is the wait and whose dead-letter target is the source queue.
The line between them is 30 seconds by default:

```python
policy = exponential_retry(6, timedelta(seconds=10))
policy.schedule()      # [10s, 20s, 40s, 80s, 160s]
policy.broker_rungs()  # [40s, 80s, 160s] — one queue each

await mq.declare(Topology().queue("shipping.orders", dead_letter=True, retry=policy))
```

`queue()` takes the **policy**, not a list of delays, because the rungs a
consumer publishes to are derived from the policy it runs: a second copy of the
list is free to drift from the first, and the way that drift shows up is a retry
addressed to a queue nobody declared, at the moment the service is already
failing.

The threshold is there because neither answer is right at both scales. A
consumer sleeping on a five-minute backoff loses the whole wait when it
restarts — the broker redelivers the unacknowledged message at once, so a
five-minute policy delivers in none. But a queue's TTL is fixed at declaration
and cannot express a jittered delay, so **jitter applies below the threshold
only**; above it the spread is free, because each message's TTL starts when it
enters the rung. Move the line with `wait_in_broker_from(...)`, and pass zero to
keep every wait in the consumer.

Per-message TTL is never used, and is the trap worth naming: RabbitMQ expires
messages only from the **head** of a queue, so one queue of per-message TTLs
lets a ten-minute wait at the front hold back every thirty-second wait behind
it. That is why a policy needs one queue per delay rather than one queue.

### The two exchanges this library declares

A rung is declared with exactly three arguments, and the same three in Java, Go,
Ruby and .NET. Two services consuming one queue declare the same rung by name, so a
rung declared with anything else answers the second service
`PRECONDITION_FAILED` and leaves it unable to consume at all — which is why the
table is pinned by a test and why `rung_args(...)` is the only place it is
written:

```python
rung_args("shipping.orders", timedelta(seconds=40))
# {'x-message-ttl': 40000,
#  'x-dead-letter-exchange': 'acemq.retry',
#  'x-dead-letter-routing-key': 'shipping.orders'}
```

| Constant | Value | What it carries |
| --- | --- | --- |
| `RETRY_EXCHANGE` | `acemq.retry` | An expired rung message, back to the queue it came from |
| `DEAD_LETTER_EXCHANGE` | `acemq.dlx` | The `{queue}.dlq` and `{queue}.parked` queues, each bound on its own name |

Both are **direct** and **durable**, and both are declared by
`Topology().queue(...)` when it is asked for retries or for dead-lettering — as
is the one binding that brings an expired message home, `{queue}` to
`acemq.retry` on `{queue}`. That binding is not optional and is not lazy: a
direct exchange drops what it cannot route and says nothing, so a topology
missing it loses every expired retry while the queue looks quiet.

Python used to dead-letter a rung through the default exchange, which needs no
exchange and no binding and works perfectly well on its own. Java has always
used the named exchange, most of the released code follows Java, and two
libraries cannot both be right about one queue. This is the settled answer, and
a broker already carrying a Java service has the arrangement above on it.

**The default policy is one delivery and no second chance.** A `retry()` on a
connection with no policy dead-letters the message and says so in the reason,
which is louder than the alternative default — an immediate requeue, which is a
hot loop nobody asked for.

### Quorum by default, and what stays classic

A durable queue asked for through `Topology().queue(...)` is a **quorum** queue,
declared with `x-queue-type: quorum`:

```python
print(Topology().queue("shipping.orders", dead_letter=True))
# Topology: 1 exchanges, 3 queues, 2 bindings
#   declare exchange acemq.dlx (direct, durable)
#   declare queue shipping.orders (durable, x-dead-letter-exchange='acemq.dlx',
#                                  x-dead-letter-routing-key='shipping.orders.dlq',
#                                  x-queue-type='quorum')
#   declare queue shipping.orders.dlq (durable)
#   declare queue shipping.orders.parked (durable)
#   declare binding shipping.orders.dlq (from acemq.dlx on shipping.orders.dlq)
#   declare binding shipping.orders.parked (from acemq.dlx on shipping.orders.parked)
```

It is the same contract as the rung arguments and it exists for the same reason.
A queue type is one more thing the broker compares, so a Java service and a
Python service consuming `shipping.orders` that disagree about it cannot both
consume it — the second one to declare is answered `PRECONDITION_FAILED`. Java
has declared quorum since it had deployments, so quorum is the answer here.

**Three kinds of queue stay classic**, deliberately, and Java declares them
classic too:

| | |
| --- | --- |
| `{queue}.retry.{delay}` | Nothing consumes a rung; replicating it buys nothing and would cost a Java service its declaration of the same name |
| `{queue}.dlq`, `{queue}.parked` | The same, for the same reason |
| Anything exclusive, auto-deleting or transient | RabbitMQ **refuses** a quorum queue that is any of those |

That last row is not a preference. A generated reply queue is exclusive and
auto-deleting so that it goes when its requester does, and asking for it as
quorum is a declaration the broker rejects — so a queue that is any of those
three is declared classic without being asked. Asking for both at once is
refused here, where the contradiction is written down, rather than at the broker,
which answers it with a message that never mentions the word quorum:

```python
Topology().queue("replies", exclusive=True, quorum=True)
# ValueError: acemq: queue 'replies' asks for quorum=True and also exclusive; ...
```

A caller who wants a classic queue says so, and gets one with no `x-queue-type`
at all — the spelling every AceMQ library uses, and therefore the only one a
broker finds equivalent to theirs:

```python
Topology().queue("shipping.orders", quorum=False)
```

Streams are unaffected: `stream(...)` writes its own `x-queue-type`, and a queue
whose arguments already name a kind keeps it.

**A queue that already exists cannot change kind.** AMQP has no way to alter a
queue's type in place, so a queue declared classic by 0.1.0 has to be drained
and deleted before this version can declare it.

### Codecs

JSON by default. A codec declares the content type it writes and says which ones
it will read, and a `CompositeCodec` tries them in order — which is what a queue
carrying two formats during a migration needs:

```python
from acemq_amqp import BytesCodec, CompositeCodec, JsonCodec, TextCodec

codec = CompositeCodec(JsonCodec(), TextCodec())
```

A message with **no** content type rules nothing out, so every codec is a
candidate and the first that can actually read the body wins. `BytesCodec`
answers for everything and hands back the bytes unchanged, which is what to read
a dead-letter queue with: the message that went there may be exactly the one
nothing could decode.

Five more formats ship behind an extra each — the same five Java and Go ship, so
a message from either is readable here:

```python
from acemq_amqp.codecs.avro import AvroCodec          # avro/binary
from acemq_amqp.codecs.protobuf import ProtobufCodec  # application/x-protobuf
from acemq_amqp.codecs.toml import TomlCodec          # application/toml
from acemq_amqp.codecs.xml import XmlCodec            # application/xml
from acemq_amqp.codecs.yaml import YamlCodec          # application/yaml

mq = await connect(url, codec=CompositeCodec(JsonCodec(), YamlCodec()))
```

Each reads a wider set than it writes, because a producer in another stack uses
whichever spelling its own library picked — `text/yaml` and `application/x-yaml`
both predate the registered `application/yaml`, and refusing them would park a
message that was perfectly readable. That is the gap these close: not
capability, interoperability. See [Codecs](docs/serialization.md), and note what
the XML codec does about document type declarations.

Avro is the one that needs a word here, because its bytes describe nothing about
themselves and a reader has to hold the schema the writer used:

```python
# the producer: registers its schema and frames the identifier into every message
codec = await AvroCodec.from_registry(registry, "order.placed", schema)

# the consumer: names its reader schema, and every message is resolved onto it —
# whichever version of the producer wrote it
codec = AvroCodec.reading(my_schema)
await codec.learn_from(registry, 7)

# a service that does both keeps one codec
codec = await AvroCodec.from_registry(
    registry, "order.placed", schema, reader_schema=my_schema
)
```

The **reader schema** is what lets a producer add a field without a synchronised
deployment: Avro is given both schemas, so a field this consumer has never heard
of is skipped and a field it expects but the producer has not started sending
arrives as the reader schema's own default. A change Avro will not resolve — a
field whose type changed, or one added without a default — is a `FatalError`
naming both schemas rather than a record of silent nonsense.

`reader_schema` is the spelling everywhere here, and the same word in the rest of
the family: Java's `readerSchema`, Ruby's `reader_schema:`, Go's `ReaderSchema`
and .NET's `ReaderSchema`. `AvroCodec.reading(...)` is that argument for a
service that only consumes — it is the one thing no constructor argument says, a
registered codec with no identifier to write.

### Encrypted bodies

`EncryptedCodec` wraps any of the above and encrypts what it produced, so the
broker, its disk, its backups and its management interface hold ciphertext:

```python
# pip install "acemq-amqp[crypto]"
from acemq_amqp.codecs.encrypted import EncryptedCodec, Keyring, generate_key

keyring = Keyring.of("2026-01", generate_key())
mq = await connect(url, codec=EncryptedCodec(JsonCodec(), keyring))
```

AES-GCM, a fresh nonce per message, and the key identifier in the clear in front
of the ciphertext — which is what makes rotation possible, because a consumer
reads which key a message needs instead of assuming the current one. The header
is bound in as associated data, so an identifier altered in flight makes the
message fail to open rather than open as something else. `key_id_of(body)`
answers "which key does this need?" from the bytes alone, without holding any.

No failure message, log line or exception ever contains the plaintext or the key,
and a wrong key and a tampered body fail identically — GCM authenticates before
it returns anything, and nothing here adds a check that would tell them apart.

**It interoperates with all four of the other AceMQ libraries.** Since the 0.5.0
round the five write one framing under that content type, byte for byte, pinned
by the same test vector in all five suites — .NET used to be the exception and no
longer is, though it alone still reads the bodies it wrote before the change. A
body framed any other way is refused here, visibly, rather than misread. The
table and the vector are in
[Codecs → Encryption](docs/serialization.md#encryption).

## Interceptors

Every organisation has something every message needs and no library can guess: a
tenant, a trace context, an authorisation token, a correlation identifier in the
logging context, a timer, a size limit. Without a seam these get copied into
every call site, where one of them is eventually forgotten and nobody finds out
until the message that needed it is the one that went without.

An interceptor is **one function handed the message and the rest of the work**:

```python
from acemq_amqp import Ack, ConsumeContext, ConsumeNext, PublishContext, PublishNext

async def stamped(context: PublishContext, send: PublishNext):
    context.set_header("tenant", tenant.current())
    return await send(context)

async def timed(context: ConsumeContext, handle: ConsumeNext) -> Ack:
    started = time.monotonic()
    try:
        return await handle(context)
    finally:
        log.info("%s took %.3fs", context.queue, time.monotonic() - started)

mq = await connect(url, on_publish=[stamped], on_consume=[timed])
# or later, which still reaches publishers that already exist:
mq.intercept_publish(stamped).intercept_consume(timed)
```

That shape rather than the pair of before-and-after hooks the Java library uses,
because Python already has the construct. `try`/`finally` runs the way out in
the reverse of the way in, nests correctly without anybody reversing a list, and
makes an interceptor that opens something and closes it **one** function instead
of two halves that have to agree. It composes the way ASGI middleware does.

Interceptors run in the order they were registered — **first registered is
outermost**, so it sees the message first on the way in and last on the way out.
That matters as soon as one reads what another wrote.

| | |
|---|---|
| `PublishContext` | `exchange`, `routing_key`, `envelope`, `payload`, `persistent`, `mandatory`, `reply_to`, and `set_header(...)`. Seen **before the codec runs**, so the payload can be changed and not only its metadata |
| `ConsumeContext` | `queue`, `envelope`, `payload`, `body`, `content_type`, `routing_key`, `redelivered`, `reply_to`, a `state` dict this library never reads, and `when_settled(...)` |

**`when_settled` is how an interceptor learns what really happened.** The `Ack`
it sees on the way out is what the handler *asked for*: a handler asking for
another attempt when there are none left is dead-lettered, and an interceptor
recording the `Ack` would report a retry that never happened. A listener
registered with `when_settled` is told the `Settlement` — `acked`, `retried`,
`rejected`, `dead_lettered` or `parked`, with the delay chosen or the reason given — once,
after the consumer decides and before it acts. It returns `False` when nothing
is driving the chain and no answer is coming.

**Refusing is raising.** A publish interceptor that raises stops the publish and
the caller sees the exception — the whole point of intercepting rather than
observing. A consume interceptor that raises is treated exactly as a handler
that raised: retried, then dead-lettered with the reason it gave. The
alternative is acknowledging a message nothing processed.

Whatever the interceptors left is what gets used: an envelope rewritten on the
way in is the one the handler is given **and** the one that gets dead-lettered,
so an operator reading `{queue}.dlq` is not missing the very field the
interceptor exists to add.

Everything an interceptor touches is public API. That is the same rule the
patterns follow, and it is what keeps this a seam rather than a privileged back
door: anything an interceptor can do, application code could have done.

The blocking API inherits them — `sync.connect(...).connection` is the
asynchronous connection underneath, and registering there applies to everything
above it.

## Telemetry and health

How much is going through, how much is failing and how long a handler takes are
numbers nobody outside the process can see. Whether the connection is really up
— as opposed to a socket that is open and wedged — is a question only something
holding the connection can ask. So the library answers both.

**No hard dependency.** The library calls a three-method `Observer` and nothing
else; a metrics client is an extra, exactly as the broker client is. Taking one
would put every user of this package on whichever was picked.

```python
from acemq_amqp import Metrics, prometheus_text

metrics = Metrics()                       # keeps the numbers in memory
mq = await connect(url, observer=metrics)
...
print(prometheus_text(metrics))           # a scrape body, nothing installed
```

```python
# Or into a registry the rest of the application already writes to:
#   pip install "acemq-amqp[prometheus]"
from acemq_amqp.prometheus import PrometheusObserver

mq = await connect(url, observer=PrometheusObserver())
```

Every name below is Java's `MetricNames`, character for character, so a
dashboard built against one library reads against another:

| Metric | |
|---|---|
| `acemq.publish.total` | Publishes. Labelled `exchange`, `routing.key` and `outcome`: `confirmed`, `unroutable`, `failed` (may have been lost: a nack, no confirm in time, an I/O error), `refused` (declined before anything was sent — a `PublishingPausedError` while the broker has blocked the connection; nothing lost, safe to retry) |
| `acemq.consume.total` | Deliveries settled. Labelled `queue` and `outcome`: `acked`, `retried`, `rejected`, `dead_lettered`, `parked` |
| `acemq.consume.duration` | Seconds, timed around the interceptors as well as the handler, carrying the same `outcome` |
| `acemq.consume.in.flight` | Being handled right now |
| `acemq.messages.retried.total` | Given another attempt. A retry says `where`: `consumer`, `broker` or `requeued` |
| `acemq.messages.dead.lettered.total` | Set aside. `outcome="dead_lettered"` went to `{queue}.dlq`; `outcome="parked"` went to `{queue}.parked` because the body would not decode or a handler returned `park(...)` |
| `acemq.retry.rung.missing` | **Worth an alert.** A long retry that had to wait in the consumer because its rung queue is not on the broker |
| `acemq.messages.set.aside.failed` | Could not be moved to a dead-letter or parking queue, so was rejected to the broker instead |
| `acemq.outbox.total` | Outbox records the relay handled. Labelled `exchange`, `routing.key` and `outcome`: `published` or `failed` |
| `acemq.outbox.lag` | **Worth an alert.** Seconds a record waited between being committed and being published — the one number that reveals a stopped relay. Java, Go and .NET write the same two |
| `acemq.request.total` | Request and reply round trips, as the caller experienced them. Labelled `routing.key` and `outcome`: `answered`, `timed_out` or `failed` |
| `acemq.request.duration` | Seconds, the same labels. A call that times out is recorded at its deadline |
| `acemq.publish.duration` | Seconds, from `send` to the broker answering, with the same labels as the publish total. Recorded for a failed publish too |
| `acemq.consume.attempts` | Which attempt a delivery was on when it arrived. Labelled `queue` |
| `acemq.pipeline.run.total` | Routing-slip runs that reached the end of their itinerary. Labelled `pipeline`, `step` and `outcome` |
| `acemq.pipeline.run.duration` | Seconds: how old the message was when it left the pipeline, so it covers the whole route rather than one hop |

**That is every name Java's `MetricNames` publishes**, so a dashboard built
against Java, Go or .NET reads against this one. Two caveats, neither of which
leaves a panel blank: `acemq.consume.attempts` is a count of attempts rather than
a number of seconds and needs whole-number histogram buckets, which
`PrometheusObserver` gives it; and `acemq.pipeline.run.total` only ever carries
`outcome="completed"`, because a routing-slip step here has no way to end a run
before its last stop. See
[observability](docs/observability.md#what-is-reported).

`routing.key` and the other dotted tag names are exported to Prometheus with
underscores, because a Prometheus label name allows nothing else.

`acemq.retry.rung.missing` is the one to alert on, because nothing else shows
it. The message is still retried and the wait still happens, so a dashboard
reads as normal — while the reason the rung exists is gone, and a consumer
restart mid-wait shortens a five-minute backoff to nothing.

### Health

```python
from acemq_amqp import BrokerHealth, aggregate_health

report = await mq.health()
report.status     # HealthStatus.UP / DOWN / DEGRADED
report.healthy    # what a readiness probe should return; degraded passes

mq.blocked        # True / False / None — has the broker blocked this connection

# Combined with the application's own checks, worst wins:
report = await aggregate_health(BrokerHealth(mq), my_database_check)
```

Two halves. The broker half asks the broker a question — a socket that is open
but wedged answers a socket-level check exactly as a healthy one does, right up
until something is asked of it. It asks one that **creates nothing**: an
exclusive queue is released only when the channel that declared it closes, so a
probe that declared one would leave a queue behind on every connection.

The consumer half is the one nothing outside can see. A consumer whose workers
have died without it being closed is one the broker is still sending messages to
and nothing is reading — indistinguishable from a quiet queue, and reported as
**degraded**: the connection works, and a replacement instance would almost
certainly stall the same way, so it is worth an alert and not worth taking out
of rotation.

**A blocked connection is up, with the reason.** RabbitMQ blocks a connection
when it is low on disk or memory and stops reading its socket until the alarm
clears. Reporting that down gets the instance restarted into the same blocked
broker, having thrown away whatever it was holding, so `parts["blocked"]` carries
the state and the status stays up — the same call Java's `AceMqHealthIndicator`
makes. `blocked` is `None` when the transport could not be asked at all, which is
a different fact from `False`. On RabbitMQ the reason is absent: the broker sends
one and aio-pika does not keep it, and an invented reason would read exactly like
one it sent.

`health()` bounds its own probe — three seconds by default — because a blocked
broker is exactly the state in which a round trip does not come back, and a
health check that hangs takes the readiness endpoint down with it.

`aggregate_health` runs checks at once rather than in turn, under a deadline, so
one that hangs cannot hang the probe with it — and a probe that hangs is a pod
that never comes back. The connection's own deadline is the shorter of the two,
so a broker that has gone quiet is described by the check that looked rather than
by the aggregate giving up on it.

### Tracing

```python
# pip install "acemq-amqp[opentelemetry]"
from acemq_amqp.tracing import OpenTelemetryTracing

OpenTelemetryTracing().install(mq)
```

Metrics answer *how much*. A trace answers *what happened to this message* — this
one was published by checkout, retried twice over four minutes and given up on —
and a counter cannot.

**A consumer's span is a child of the publish that caused it**, taken from the
message's own headers rather than from whatever context happened to be current
when the delivery arrived. Those are different processes and often minutes apart,
and joining them is the one thing a messaging system needs from tracing that an
HTTP client does not.

The context travels in `traceparent` and `tracestate` — deliberately **not**
`x-acemq-` prefixed, unlike every other header here, because they are the W3C
names every other piece of tracing tooling already reads. Java, Go and .NET write
the same two.

| Span | Kind |
|---|---|
| `<destination> publish` | `PRODUCER` |
| `<queue> process` | `CONSUMER` |
| `<destination> request` | `CLIENT` — because that one *waits*, so its duration measures a responder rather than a broker |

`unroutable`, `failed` and `dead_lettered` set the span status to `ERROR`; the
others, `retried` and `parked` included, do not (a `refused` publish is red by
its exception, not its outcome) — a retry is the system working,
a park is a decision a handler made on purpose, and a wall of red traces that
turned out fine is how people learn to ignore the colour.

**The outcome is the consumer's decision, not the handler's answer**, and so is
the counter beside it. A `process` span stays open past the handler and takes
its outcome from what the consumer actually did, so a message that ran out of
attempts reads `dead_lettered` and carries a `message.dead_lettered` event with
the reason — rather than `retried`, which is what somebody querying for dead
letters finds nothing under. `acemq.messages.dead.lettered.total` is incremented
for the same delivery, and `acemq.messages.retried.total` is not; the counters are chosen
from the same settlement, and a test holds the two renderings together. A retry
carries `message.retried` with the delay the policy chose, which is a jittered
number that exists nowhere else. The backoff itself is not inside the span.

A `request` span ends `answered` or `timed_out` — Java's two words for the same
span — so a round trip that worked has an outcome to count and not only the ones
that did not. `messaging.rabbitmq.destination.routing_key` is a publish-side
attribute only, as it is everywhere else.

Spans are recorded under the instrumentation scope `org.acemq.amqp`, which is
what Java, Ruby, Go and .NET register too, so one query reads across all five.

The dependency is `opentelemetry-api`, not the SDK, so without an application
configuring one this exports nothing at all. See
[Tracing](docs/observability.md#tracing).

## Patterns

`acemq_amqp.patterns` holds the things every service that consumes a queue ends
up writing for itself. They mean the same as the Go library's, because a routing
slip written by a Go service has to be readable by a Python one; the API shape is
Python's.

```python
from acemq_amqp.patterns import InMemoryIdempotencyStore, chain, idempotent, with_timeout
```

| | |
|---|---|
| `idempotent(store, handler)` | Handle a message once however many times it arrives. A duplicate is **accepted**, not rejected: the work was done |
| `record(...)` / `OutboxRelay` | Write the message into the same transaction as the work, and let a relay publish what was committed |
| `Requester` / `serve(...)` | Ask a question and wait for the answer. A responder's failure comes back as a failure, not as a timeout. The return address is written to both the `acemq-reply-to` header and AMQP's own `reply-to` property, and read header-first, so any of the five can call any other. Both halves count what they did — `answered`, `unanswerable`, `timed_out`, `unmatched` — and the caller's round trip is timed on `acemq.request.duration` |
| `replay(...)` | Put dead letters back, with a filter, a limit and a deadline, and a report of what it did and why it stopped |
| `ordered(key, handler)` | Keep one entity's messages in sequence while everything else runs at once |
| `ConsumerGroup` | Several consumers over one queue, started and stopped as one thing |
| `RoutingSlip` / `follow_slip(...)` | An itinerary the message carries, instead of an orchestrator that knows it. Both wire forms are read — the JSON slip Python, Go and Ruby write, and the declared `x-acemq-route` Java's `Pipeline` writes — and either can be written |
| `Saga` | Steps that must all happen across systems sharing no transaction, compensated in reverse when one fails. It reports what could **not** be undone rather than raising |
| `Scheduler` | Deliver a message later, through a ladder of time-to-live queues rather than a per-message expiration that a queue only honours at its head |
| `chain(...)` / `then(...)` | Wrap a handler in a deadline, logging and the guards; publish what a step produced onwards |
| `SchemaRegistry` | Remember what a message used to look like, so a producer can add a field without a synchronised deployment |
| `ClaimCheckCodec` | Put a payload too large for a broker into a store and send the key, above a threshold and inline below it |
| `read_stream(...)` | Read a queue that keeps what it has already handed out |

Every one is built out of the public library — handlers, envelopes, publishers —
so nothing is possible with a pattern that would not be possible without it.

### Stores that survive a restart

Four of these have a storage seam, and each in-memory implementation says in its
own docstring why it is not the one to use in production. Three of them —
`IdempotencyStore`, `OutboxStore` and `SchemaRegistry` — are backed by
`acemq_amqp.patterns.sql`:

```python
import sqlite3

from acemq_amqp.patterns import SqlOutboxStore, create_schema

connections = lambda: sqlite3.connect("acemq.db")   # or a pool's checkout
create_schema(connections)                          # development only
outbox = SqlOutboxStore(connections)

async with database.transaction() as tx:
    await place_order(tx, order)
    await outbox.add(record(mq, "orders-events", "order.placed", event),
                     connection=tx.connection)
# one commit; the order and the message are the same decision
```

`add` writes on the connection **you** hand it and does not commit it, does not
roll it back and does not close it. That is the guarantee rather than an
oversight: roll your transaction back and the message is not in the outbox,
because it never was. With no transaction to join it raises rather than opening
one — a fresh connection with autocommit on would leave a message queued for
work that never happened, which is the exact fault the pattern was adopted to
prevent.

The fourth is `ClaimCheckStore`, which is not in the SQL module and is
synchronous rather than awaitable, because a `Codec` is called from inside both
the async and the blocking API. `FilesystemClaimCheckStore` ships; the protocol
is `put`, `get` and `delete`, so a store over S3 or a blob container is a small
class.

Nothing in the SQL module imports a database driver. It is written against the
DB-API 2.0 protocols, so `sqlite3` from the standard library works with nothing
installed and psycopg works if you have it — `paramstyle="format"` for the
latter. **The automated suite exercises `sqlite3`**; the same checks have been
run by hand against PostgreSQL 17 through psycopg 3 and pass, but they are not
in the suite, because a suite that needs a database server is a suite that gets
skipped. Anything else — MySQL, SQL Server, Oracle — needs at least a different
upsert and is not claimed.

## Requirements

Python 3.10 or newer, and RabbitMQ for the transport — **3.13 or 4.x**. CI runs
the whole broker suite against both on every push, so neither is an assumption.

## Development

```bash
python3 -m venv .venv && . .venv/bin/activate
pip install -e ".[dev]"
pytest -m "not integration"     # the contract and the engine, no broker needed
ruff check . && mypy

# And against a real one:
ACEMQ_TEST_BROKER=amqp://guest:guest@localhost:5672/ pytest -m integration
```

The TLS tests need a broker with a TLS listener, and skip when they are not told
where one is. Two are wanted, because one broker cannot both accept a client
that has no certificate and refuse it:

```bash
export ACEMQ_TEST_TLS_CERTIFICATES=/path/to/certs   # ca.crt, client.crt, client.key,
                                                    # other-ca.crt, stranger.crt, stranger.key
export ACEMQ_TEST_TLS_BROKER=amqps://localhost:25891/          # ssl_options.verify = verify_none
export ACEMQ_TEST_TLS_MUTUAL_BROKER=amqps://localhost:25893/   # verify_peer, fail_if_no_peer_cert
pytest -m integration
```

They prove the handshake rather than the configuration: the certificate the
broker presented, the version and cipher that were agreed, that a broker the
system trust store does not vouch for is **refused**, and that a client
certificate signed by the wrong authority gets no further than one signed by
none.

The fixtures under `tests/fixtures/` are generated by the Java implementation
and committed byte for byte by all five libraries.
`envelope-fixtures.json` is the wire contract — the headers that travel with a
message. `contract-fixtures.json` is the behaviour contract: the retry
schedules, the jitter bounds, where a wait is spent, the queue names, the rung
arguments, the declared topology and the queue types. Both are checked here
rather than assumed, and `../scripts/check-fixtures.sh` compares every library's
copy so that a file edited in one repository cannot quietly become that
repository's private opinion.

The integration tests name everything they create `pyit.*` and delete it
afterwards, so they can be pointed at a broker that is not theirs alone.

### The documentation site

```bash
pip install -e ".[dev]"
bash .github/scripts/build-docs-site.sh    # needs pandoc as well
open site/index.html
```

The prose plus an API reference under `/apidocs/`, generated by pdoc from the
docstrings in `src/`. Go links out to pkg.go.dev instead, because Go has one
built from the source of every published version; Python has nothing equivalent
for a package, so nothing else will host this. Generating it from the same
commit as the prose is what stops the two drifting apart.

`docs.yml` publishes it to GitHub Pages on a push to `main`, and fails the build
on a link to a page that does not exist or an API reference missing a module —
pdoc reports a module it cannot import as a page rather than as an error, so a
missing dependency otherwise produces a reference that is present, linked and
empty.

### Releasing

```bash
git tag v0.7.0 && git push origin v0.7.0
```

`release.yml` runs the whole suite including the integration tests, builds the
sdist and the wheel, checks that the sdist alone installs and imports, uploads
to PyPI, then installs the published version from PyPI into an empty
interpreter to confirm consumers get what was built.

It publishes through **PyPI Trusted Publishing** — the runner mints a
short-lived OIDC token for this workflow and this environment, and PyPI checks
it. No upload token is stored. That matters more here than for a feed we
control: a leaked token can publish any version to an index every Python
installation in the world resolves from, and a published version is permanent.

Two guards, both copied from the Go, Java and .NET release workflows. The
version must be a version — an allow-list, because it reaches a file name and a
package version — and it must be `0.7.x`, so a mistyped tag cannot ship a `1.0.0`
that nobody can withdraw. The tag is what names the release, and `release.yml`
refuses to publish unless `pyproject.toml` and `acemq_amqp.__version__` both
already say the same thing, so the version in the tree is bumped before the tag
is cut rather than after.

Before the first tag, PyPI needs a pending publisher for `acemq-amqp` naming
this repository, `release.yml` and the `pypi` environment, and this repository
needs that environment. Until then a tag reaches the upload and PyPI refuses it,
which is the right way round: nothing is published by accident and the failure
says what is missing.

## Licence

Apache-2.0. RabbitMQ is a trademark of Broadcom Inc.; this project is not
affiliated with it.
