aioworkers-mqtt 0.1.0

aioworkers plugin integration to mqtt

Features

• Queue over mqtt

• Storage over mqtt

Install

Use uv to add to your project (recommend):

uv add aioworkers-mqtt[paho]

or install with pip:

pip install aioworkers-mqtt[paho]

Usage

Add this to aioworkers config.yaml:

queue:
  cls: aioworkers_mqtt.Queue
  host: localhost
  port: 1883
  user: user
  password: ***
  qos: 1
  format: json
  topics:
    - a
    - b

storage:
  cls: aioworkers_mqtt.Storage
  host: localhost
  port: 1883
  user: user
  password: ***
  qos: 1
  retain: true
  format: json
  prefix: room/device

You can work with queue like this:

await context.queue.put({"a": 1})
d = await context.queue.get()

and work with storage like this:

await context.storage.set("a", 1)
one = await context.storage.get("a")

Run:

aioworkers -c config.yaml -l debug

Development

Check code:

hatch run lint:all

Format code:

hatch run lint:fmt

Run tests:

hatch run pytest

Run tests with coverage:

hatch run cov