Metadata-Version: 2.4
Name: mqttium
Version: 0.1.0a4
Summary: Reliable async-native MQTT client for Python
Project-URL: Homepage, https://github.com/yoch/mqttium
Project-URL: Repository, https://github.com/yoch/mqttium
Project-URL: Issues, https://github.com/yoch/mqttium/issues
Project-URL: Changelog, https://github.com/yoch/mqttium/blob/main/CHANGELOG.md
Author-email: Yoch Melka <795960+yoch@users.noreply.github.com>
License-Expression: Apache-2.0
License-File: LICENSE
License-File: NOTICE
Keywords: asyncio,iot,messaging,mqtt
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Communications
Classifier: Topic :: Internet
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Provides-Extra: dev
Requires-Dist: mypy>=1.17; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest-cov>=6; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.16; extra == 'dev'
Provides-Extra: fuzz
Requires-Dist: hypothesis>=6.100; extra == 'fuzz'
Requires-Dist: pytest>=8; extra == 'fuzz'
Provides-Extra: release
Requires-Dist: build>=1.2; extra == 'release'
Requires-Dist: check-wheel-contents>=0.6; extra == 'release'
Requires-Dist: twine>=6; extra == 'release'
Requires-Dist: validate-pyproject[all]>=0.24; extra == 'release'
Provides-Extra: security
Requires-Dist: bandit>=1.8; extra == 'security'
Description-Content-Type: text/markdown

# MQTTium

**MQTTium is a reliable, async-native MQTT client for modern Python.**

It combines a synchronous protocol engine with an `asyncio` API, bounded
backpressure, durable inflight persistence, explicit delivery receipts, and
complete MQTT QoS state machines.

> **Status:** alpha (`0.1.0a4`). The implementation is extensively tested, but
> the public API may still change before the first stable release.

## Features

- MQTT 3.1.1 and MQTT 5;
- QoS 0, 1 and 2 with one authoritative protocol state machine;
- TCP, TLS, WebSocket and Unix transports;
- reconnect and session replay;
- bounded callback and async-iterator delivery;
- immutable runtime statistics for queues, budgets, receipts and transports;
- manual acknowledgement;
- in-memory and SQLite inflight persistence;
- aggregate `publish_many()` with bounded memory and measured throughput gains;
- synchronous loop-bound `publish_nowait()` for non-suspending native producers;
- an additive Paho VERSION2 compatibility façade, isolated from the native API;
- inline type information for type checkers.

Python **3.11–3.14** is supported. MQTTium is licensed under **Apache-2.0**.

## Installation

```bash
python -m pip install mqttium
```

To work on the unreleased development version:

```bash
git clone https://github.com/yoch/mqttium.git
cd mqttium
python -m pip install -e ".[dev,fuzz,release]"
```

## Quick start

```python
import asyncio

from mqttium.api import AsyncClient


async def main() -> None:
    client = AsyncClient("demo-client")
    await client.connect("127.0.0.1", 1883)
    await client.subscribe("demo/#")

    receipt = await client.publish("demo/hello", b"world", qos=1)
    await receipt.wait()

    await client.disconnect()


asyncio.run(main())
```

## Non-suspending publishing

A producer already executing on the client's event loop can submit without
creating or awaiting a coroutine:

```python
receipt = client.publish_nowait("telemetry/device-1", payload, qos=1)
# Continue synchronously, then observe completion later if needed.
await receipt.wait()
```

`publish_nowait()` either admits the publication immediately or raises
`FlowControlError`; it never waits for engine or writer capacity. It returns the
normal `PublishReceipt`, so QoS 1/2 completion is observed in the same way as
with `publish()`.

The method follows the same ownership rule as `asyncio.Queue.put_nowait()`: it
is intended for the client's owning event-loop thread, not as a generic
thread-safe API. Cross-thread synchronous callers should use an adapter such as
`mqttium.compat.paho.Client`, which coalesces submissions before handing a
bounded batch to the loop.

## Batched publishing

`publish_many()` consumes iterables in bounded chunks and returns one aggregate
receipt rather than creating one task or event per message:

```python
from mqttium.api import PublishMessage

batch = await client.publish_many(
    PublishMessage("telemetry/device-1", payload, qos=1)
    for payload in payloads
)
await batch.wait()
```

The retained paired A/B benchmark measured publisher-throughput geomean
improvements of **36.9% for QoS 0**, **15.9% for QoS 1**, and **7.0% for QoS 2**
against equivalent individual publishing pipelines on the validated source
tree. Benchmark methodology and limitations are documented in
[`docs/BENCHMARKING.md`](docs/BENCHMARKING.md).

## Bounded memory

Every queue that can grow with application load is bounded by default, so a
producer that outruns its broker is slowed down rather than allowed to exhaust
the process:

```python
client = AsyncClient(
    max_pending_outbound_messages=10_000,   # unfinished QoS 1/2 publications
    max_pending_outbound_bytes=64 * 1024**2,  # their logical topic+payload+properties
    max_pending_delivery_bytes=64 * 1024**2,  # inbound messages awaiting a consumer
    max_ingress_batch_bytes=1 * 1024**2,      # decoded work before delivery is drained
    publish_backpressure="wait",             # or "error" to refuse immediately
)
```

`publish()` waits for capacity by default and raises `FlowControlError` under
`publish_backpressure="error"` or with `nowait=True`. A refusal is atomic: no
packet identifier is allocated and no store record is written. Pass `None` for
any limit to restore unbounded queueing.

These defaults are new in `0.1.0a2`; before them a QoS 1/2 producer could queue
until the 65 535 packet-identifier space was exhausted. See
[`docs/MIGRATION.md`](docs/MIGRATION.md).

## Runtime statistics

`stats()` returns a frozen snapshot without starting a sampler or emitting logs:

```python
snapshot = client.stats()
print(snapshot.outbound.pending_bytes)
print(snapshot.inbound.inflight)
print(snapshot.writer.queued_bytes)
print(snapshot.delivery.pending_bytes)
```

Each section is produced by the component that owns the state — the two protocol
sessions, the effect and write pumps, the transport — and `stats()` only
assembles them. The snapshot also includes lifetime high-water marks, batching
decision counters, task state, receipt counts, decoder buffering and
WebSocket/stream transport buffers. It is intended to be called on the client's
owning event loop. See [`docs/API-STABILITY.md`](docs/API-STABILITY.md).

## Validation

The release gates include:

- more than 300 unit tests;
- Mosquitto integration tests on Python 3.11, 3.12, 3.13 and 3.14;
- deterministic and Hypothesis-based fuzzing;
- Ruff formatting and linting;
- mypy validation and a PEP 561 `py.typed` marker;
- an 80% coverage gate;
- wheel, source-distribution and isolated-install validation;
- delivery, persistence, TCP, TLS and WAN-profile benchmarks.

A separate finalisation workflow runs short reconnect/backpressure soaks on
Linux for relevant pull requests. Extended Linux/macOS soaks and interoperability
campaigns against multiple brokers are available by manual dispatch. Their
acceptance criteria are documented in [`docs/STABILITY.md`](docs/STABILITY.md).

## Documentation

- [`docs/DESIGN.md`](docs/DESIGN.md) — architecture and invariants
- [`docs/API-STABILITY.md`](docs/API-STABILITY.md) — public API candidate and deprecations
- [`docs/STABILITY.md`](docs/STABILITY.md) — soak and interoperability campaign
- [`docs/IMPLEMENTATION-GUIDE.md`](docs/IMPLEMENTATION-GUIDE.md) — protocol contracts
- [`docs/COMPAT.md`](docs/COMPAT.md) — Paho compatibility surface
- [`docs/MIGRATION.md`](docs/MIGRATION.md) — migration guidance
- [`docs/BENCHMARKING.md`](docs/BENCHMARKING.md) — benchmark validity contract
- [`docs/FUZZING.md`](docs/FUZZING.md) — fuzzing strategy
- [`docs/ROADMAP.md`](docs/ROADMAP.md) — remaining stable-release work
- [`PROVENANCE.md`](PROVENANCE.md) — source history and licensing review

## Contributing and security

See [`CONTRIBUTING.md`](CONTRIBUTING.md). Report vulnerabilities through the
private process described in [`SECURITY.md`](SECURITY.md).
