DOCSEARCHbyAlgolia
Submit Docs

aiomqtt Documentation

repository·main·Indexed 20 days ago

https://github.com/empicano/aiomqtt

An idiomatic asyncio-based MQTT client for Python providing a modern interface for MQTTv5. It replaces traditional callback patterns with asynchronous context managers and iterators. Key features include full MQTTv5 support, automatic reconnection, type safety, and a pure asyncio implementation in version 3.0.0-alpha.1.

Tokens
13.3K
Snippets
38
Records
49
Agent score
69%

What's inside aiomqtt

  1. Overview of aiomqtt features

    main

    aiomqtt is an idiomatic asyncio MQTT client designed for modern Python development. Key characteristics include:

    • No Callbacks: Uses asynchronous iteration and context managers instead of traditional callback functions.
    • MQTTv5 Support: Full support for MQTTv5 features like backpressure and user properties.
    • Automatic Reconnection: Handles connection drops automatically.
    • Pure asyncio: (In v3 alpha) The library is entirely asyncio-based with no reliance on threads.
    • Type Safety: Fully type-hinted for better developer experience.

  2. Understand MQTT Quality of Service (QoS) levels

    main

    MQTT supports three levels of reliability for message delivery:

    • 0 (At most once): The fastest option. The message is sent once with no guarantee of delivery.
    • 1 (At least once): The message is delivered at least once. The receiver acknowledges receipt with a PUBACK packet.
    • 2 (Exactly once): The most reliable but slowest option. It uses a four-part handshake (PUBLISH, PUBREC, PUBREL, PUBCOMP) to guarantee exactly one delivery.

    Note on QoS: Since MQTT typically runs over TCP, QoS is primarily relevant during reconnections (retransmitting unacknowledged messages when clean_start=False) and for flow control. QoS=0 messages may be dropped under high load.

  3. Configure persistent sessions to prevent message loss

    main

    By default, brokers discard a client's session (subscriptions and unacknowledged QoS=1/2 messages) upon disconnection. To maintain a persistent session so the broker queues messages while the client is offline, configure the following:

    1. Set clean_start=False.
    2. Set a non-zero session_expiry_interval (in seconds).

    Tip: To create a session that never expires, set session_expiry_interval to 2**32 - 1.

  4. Use retained messages

    main

    A retained message is a message published with retain=True. The broker stores the most recent retained message for a specific topic. When a new client subscribes to that topic, the broker immediately sends them that stored message.

    Key behaviors:

    • The broker stores only one retained message per topic.
    • Publishing a new retained message to the same topic overwrites the previous one.
    • To delete a retained message for a topic, publish a retained message with an empty payload.

  5. Manage the Client connection lifecycle

    main

    The aiomqtt.Client is designed to be used as an asynchronous context manager.

    • Entering the context: The client sends a CONNECT packet. The __aenter__ method returns only after the broker responds with CONNACK.
    • Exiting the context: The client sends a DISCONNECT packet and closes the connection.

    Manual Lifecycle Management (Workaround): If you cannot use a context manager, you can manually call await client.__aenter__() and await client.__aexit__(...). If you use this approach, you must ensure __aexit__ is called even if exceptions occur to avoid leaking connections.

  6. Configure subscription QoS with max_qos

    main

    When subscribing, you can set the max_qos parameter in aiomqtt.TopicFilter to control how the broker delivers messages to your client:

    • QoS.AT_MOST_ONCE (0): The broker downgrades all messages to QoS 0.
    • QoS.AT_LEAST_ONCE (1): The broker downgrades QoS 2 messages to QoS 1.
    • QoS.EXACTLY_ONCE (2, default): Messages are delivered at the QoS level with which they were published.

  7. Apply backpressure using receive_max

    main

    You can prevent a broker from overwhelming your client by setting the receive_max parameter in the aiomqtt.Client constructor.

    When the number of unacknowledged QoS=1 and QoS=2 messages reaches the receive_max limit, the broker will queue further messages.

    Note: Backpressure does not apply to QoS=0 messages; they are sent regardless of the receive_max setting.

  8. Implement a message distributor for multiple queues

    main

    Since aiomqtt v2.0.0 uses a single client-wide message queue, you can no longer create isolated queues via method calls. If your application requires separate queues for different topics (e.g., for specific concurrency requirements), you must implement a "distributor" pattern.

    Pattern:

    1. Create multiple asyncio.Queue objects.
    2. Create a distributor task that iterates over client.messages.
    3. Use message.topic.matches() within the distributor to route messages to the appropriate asyncio.Queue using put_nowait().
    4. Run the distributor and your topic-specific consumers concurrently using an asyncio.TaskGroup.
    import asyncio
import aiomqtt

async def temperature_consumer(queue):
    while True:
        message = await queue.get()
        print(f"[temperature/#] {message.payload}")

async def humidity_consumer(queue):
    while True:
        message = await queue.get()
        print(f"[humidity/#] {message.payload}")

temperature_queue = asyncio.Queue()
humidity_queue = asyncio.Queue()

async def distributor(client):
    async for message in client.messages:
        if message.topic.matches("temperature/#"):
            temperature_queue.put_nowait(message)
        elif message.topic.matches("humidity/#"):
            humidity_queue.put_nowait(message)

async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        await client.subscribe("temperature/#")
        await client.subscribe("humidity/#")
        async with asyncio.TaskGroup() as tg:
            tg.create_task(distributor(client))
            tg.create_task(temperature_consumer(temperature_queue))
            tg.create_task(humidity_consumer(humidity_queue))

asyncio.run(main())

  9. Run a message listener without blocking

    main

    Since async for message in client.messages() is an infinite loop, you should run the consumer in a separate task using asyncio.create_task() or asyncio.TaskGroup to allow other code to execute concurrently.

    import asyncio
import aiomqtt


async def consume(client: aiomqtt.Client) -> None:
    async for message in client.messages():
        print(message.payload)


async def main() -> None:
    async with aiomqtt.Client(hostname="test.mosquitto.org") as client:
        await client.subscribe(aiomqtt.TopicFilter("ducks/#", max_qos=aiomqtt.QoS.AT_MOST_ONCE))
        async with asyncio.TaskGroup() as tg:
            tg.create_task(consume(client))
            tg.create_task(asyncio.sleep(5))  # Some other task


asyncio.run(main())

  10. Migrate from v2 to v3: Publishing Messages

    main

    In v3, payloads must be explicitly provided as bytes. Passing str, int, or float is no longer supported. Additionally, publishing with QoS > 0 now requires a packet_id to support retry patterns.

    Publishing with QoS > 0: You must obtain a packet_id from the client's packet_ids generator.

    # v2
await client.publish("topic", payload="hello")
await client.publish("topic", payload=42)
await client.publish("topic", payload=b"hello", qos=1)

# v3
await client.publish("topic", b"hello")
await client.publish("topic", b"42")

# v3 with QoS > 0
packet_id = next(client.packet_ids)
await client.publish(
    "topic", b"hello", qos=aiomqtt.QoS.AT_LEAST_ONCE, packet_id=packet_id
)

  11. Migrate client lifecycle management in aiomqtt v2.0.0+

    main

    In aiomqtt v2.0.0 and later, the connect and disconnect methods have been removed.

    Recommended: Use a Context Manager

    The preferred way to manage the connection lifecycle is using the Client as an asynchronous context manager. This ensures the client connects upon entry and disconnects upon exit.

    Alternative: Manual Lifecycle Management

    If you cannot use an async with block, you must manually call the __aenter__ and __aexit__ methods:

    • await client.__aenter__() replaces connect.
    • await client.__aexit__(None, None, None) replaces disconnect. Note that __aexit__ expects three arguments (exc_type, exc, and tb) representing the exception context; pass None for all three when calling manually.
    import asyncio
import aiomqtt

# Recommended approach
async def main():
    async with aiomqtt.Client("test.mosquitto.org") as client:
        await client.publish("temperature/outside", payload=28.4)

# Manual approach
async def main_manual():
    client = aiomqtt.Client("test.mosquitto.org")
    await client.__aenter__()
    try:
        await client.publish("temperature/outside", payload=28.4)
    finally:
        await client.__aexit__(None, None, None)

asyncio.run(main())

  12. Handle messages concurrently with multiple tasks

    main

    The Client.messages() generator yields messages sequentially. If message processing is I/O-bound (e.g., writing to a database), you can increase throughput by spawning multiple consumer tasks using an asyncio.TaskGroup.

    Important Considerations:

    • This approach is only beneficial for I/O-bound code. For CPU-bound tasks, use multiple processes.
    • Messages might be processed in a different order than they arrived.
    import asyncio
import aiomqtt
import random


async def consume(client: aiomqtt.Client) -> None:
    async for message in client.messages():
        await asyncio.sleep(random.random())  # Simulate some I/O-bound work
        print(message.payload)


async def main() -> None:
    async with aiomqtt.Client("test.mosquitto.org", receive_max=16) as client:
        await client.subscribe(aiomqtt.TopicFilter("ducks/#", max_qos=aiomqtt.QoS.AT_MOST_ONCE))
        async with asyncio.TaskGroup() as tg:
            for _ in range(4):  # The number of consumer tasks
                tg.create_task(consume(client))


asyncio.run(main())