django-mqtt-client 0.1.1

Asynchronous MQTT client library for Python with optional Django integration

django-mqtt-client

Unified inter-service event bus utility for Django services (MQTT + RabbitMQ).

Objectives

• Standardize inter-service messaging using one RabbitMQ envelope schema.
• Support request/response/event patterns with topic conventions.
• Ensure reliable publish/consume with retries, backoff, and worker heartbeats.
• Persist envelope metadata in DB (message_type, message_status, source_service, reply_to, parent_request_id, chain_depth).
• Guarantee message identity with message_id (auto-generated/fallback from OutboundEvent.id when missing on outbound).
• Keep MQTT inbound compatible for EdgeX events while enforcing schema for RabbitMQ.

What it does

• Outbound flow: DB OutboundEvent -> broker publish
• Inbound flow: broker consume -> DB InboundEvent
• Shared RabbitMQ envelope validation via RabbitMqSchemaValidator
• Retry + backoff for outbound publish failures
• Worker heartbeat/status updates via runtime config table

Install

pip install django-mqtt-client

Django setup

INSTALLED_APPS = [
    # ...
    "django_mqtt_client",
]

Run migrations:

python manage.py migrate

Key modules

• django_mqtt_client/event_bus.py
  • async worker for MQTT and RabbitMQ
• django_mqtt_client/rabbitmq_schema.py
  • envelope normalize/validate for RabbitMQ messages
• django_mqtt_client/exceptions.py
  • broker-neutral exceptions

RabbitMQ envelope

Schema: interservice.envelope.v1

{
  "schema": "interservice.envelope.v1",
  "message_id": "uuid",
  "message_type": "request",
  "topic": "edgenexus.request.machine.read",
  "source_service": "ebmr",
  "request_id": "req-100",
  "parent_request_id": "req-99",
  "chain_depth": 1,
  "reply_to": "ebmr.response.machine.read",
  "payload": {
    "machine_id": "M1"
  }
}

Message type rules

• request: requires request_id, reply_to
• response: requires request_id, status (success or error)
• event: requires event_type

message_id behavior

• Envelope includes message_id.
• Outbound RabbitMQ: if message_id is missing, utility auto-generates a UUID.
• Outbound fallback: if DB row has no envelope message_id, OutboundEvent.id is used before validation.
• Inbound RabbitMQ: message_id must be present after normalization; invalid messages are dropped.

Topic rules

• Request: <target_service>.request.<entity>.<action>[.<subaction>...]
• Response: <requesting_service>.response.<entity>.<action>[.<subaction>...]
• Event: event.<domain>.<entity>.<action>[.<subaction>...]

Examples:

• edgenexus.request.machine.read
• ebmr.response.machine.read
• event.operational.machine.status.changed

Message samples

Request:

{
  "message_type": "request",
  "topic": "edgenexus.request.machine.read",
  "request_id": "req-100",
  "reply_to": "ebmr.response.machine.read",
  "payload": {
    "machine_id": "M1"
  }
}

Response:

{
  "message_type": "response",
  "topic": "ebmr.response.machine.read",
  "request_id": "req-100",
  "status": "success",
  "payload": {
    "machine_id": "M1",
    "state": "running"
  }
}

Event:

{
  "message_type": "event",
  "topic": "event.operational.machine.status.changed",
  "event_type": "machine.status.changed",
  "payload": {
    "machine_id": "M1",
    "status": "running"
  }
}

Chaining sample:

{
  "message_type": "request",
  "topic": "edgenexus.request.batch.start",
  "request_id": "req-200",
  "parent_request_id": "req-100",
  "chain_depth": 1,
  "reply_to": "ebmr.response.batch.start",
  "payload": {
    "batch_id": "B-001"
  }
}

Runtime DB config

Worker reads active row from EventBusRuntimeConfig (config_key, is_active).

Broker-specific child settings:

• EventBusMqttSettings
• EventBusRabbitMqSettings

Shared connection fields are in runtime config (host, port, username, password, connect_timeout, retry_interval_seconds).

Event table fields

OutboundEvent

• transport fields:
  • status (pending/published/invalid/error) for publish lifecycle
  • published, published_at, retry_count, next_retry_at, last_error
• envelope-related fields:
  • topic
  • request_id
  • event_type
  • message_type
  • message_status (mapped to envelope status)
  • source_service
  • parent_request_id
  • chain_depth
  • reply_to
  • payload
  • id is used as fallback message_id if payload/message row does not provide one

InboundEvent

• topic
• request_id
• message_type
• message_status (envelope/business status if present)
• reply_to
• source_service
• parent_request_id
• chain_depth
• payload
• processing fields:
  • status (pending/processed/error) for local processing lifecycle
  • processed, processed_at, last_error

EVENT_BUS_CONFIG example

Worker

Start:

python manage.py event_bus_worker --config-key default

Optional override:

python manage.py event_bus_worker --config-key default --broker-type rabbitmq

Status:

python manage.py event_bus_status

Behavior notes

• Broker connect failure retries every configured retry interval.
• Outbound invalid payload/envelope is marked status=invalid.
• Outbound publish errors use exponential backoff until max_retry_count.
• If InboundEvent.message_id exists, duplicate inbound message IDs are ignored.

Tests

Run from repo root:

pytest -q

If your environment does not include coverage plugins from pyproject.toml, run:

pytest -q -o addopts=''