pyqmh

Python Queued Message Handler (pyqmh) is a lightweight in-process messaging framework for coordinating modules with request/response, actions, and broadcast messaging.

Installation

From PyPI:

pip install pyqmh

From source:

pip install .

Quick Start

from pyqmh import Module, Protocol


class Worker(Module):
 def handle_message(self, message):
  if message.command == "ping":
   self.protocol.send_response(message, "pong")
   return False
  return super().handle_message(message)


protocol = Protocol(main_address="main")
worker = Worker("worker", protocol)

response = protocol.send_request("worker", "ping", timeout=1.0)
print(response)

protocol.send_action("worker", "shutdown")

Debugging and Logging

Module uses Python logging instead of ad-hoc debug prints. Each module instance gets a logger named by address (for example: pyqmh.module.worker).

Enable debug output

import logging

logging.basicConfig(
 level=logging.INFO,
 format="%(asctime)s %(levelname)s %(name)s - %(message)s",
)

Worker example with custom debug logs

from pyqmh import Module, Protocol


class Worker(Module):
 def handle_message(self, message):
  self.logger.debug(
   "Received command=%s payload=%r source=%s",
   message.command,
   message.payload,
   "yes" if message.source is not None else "no",
  )

  if message.command == "ping":
   self.logger.debug("Responding with pong")
   self.protocol.send_response(message, "pong")
   return False

  return super().handle_message(message)


protocol = Protocol(main_address="main")
worker = Worker("worker", protocol, debug=True)

response = protocol.send_request("worker", "ping", payload={"id": 123}, timeout=1.0)
print(response)

protocol.send_action("worker", "shutdown")

Tips:

• Pass debug=True when creating a module to force that module logger to DEBUG.
• Pass logger=<your logger> to Module(...) to attach modules to your app logger hierarchy.
• In child classes, use self.logger.debug(...) for diagnostic output.

Public API

The package exports the following symbols at top-level:

• pyqmh.Protocol
• pyqmh.ProtocolError
• pyqmh.Module
• pyqmh.Message

Testing

Install the package with test dependencies:

pip install -e ".[test]"

Run the full test suite:

pytest tests/ -v

Run tests with coverage report:

pytest tests/ --cov=src/pyqmh --cov-report=html

Run only unit tests or integration tests:

pytest tests/ -m unit      # unit tests only
pytest tests/ -m integration  # integration tests only

Test Coverage

The test suite includes:

• 81 tests across Message, Protocol, and Module classes
• 100% code coverage of the package
• Unit tests for core functionality (instantiation, equality, hashing, message handling)
• Integration tests for multi-module communication and broadcast messaging
• Thread-safe test cleanup to ensure clean pytest exit without requiring keyboard interrupt

Test files:

• tests/test_message.py - Message class tests (creation, equality, hashing, payloads)
• tests/test_protocol.py - Protocol class tests (registration, message sending/receiving, requests/responses)
• tests/test_module.py - Module class tests (threading, message handling, background tasks, cleanup)

Development Build

Build source and wheel distributions:

python -m build

Validate metadata and artifacts:

twine check dist/*