Skip to content

blackbull.mqtt.asyncapi

blackbull.mqtt.asyncapi

AsyncAPI 3.0 documentation for registered application taps.

This describes opaque tap payloads, not broker QoS or retained-delivery guarantees. Register alongside MQTTExtension through app.add_extension.

AsyncAPIExtension

Bases: Extension

Mount an AsyncAPI 3.0 spec endpoint (+ HTML viewer) describing the MQTT taps.

Requires an MQTTExtension on the same app; it is read lazily (per request) via app.extensions['mqtt'], so there is no ordering constraint at registration — and taps added after this extension are still documented. The MQTT extension must, however, be present when the spec route is hit, or the request raises RuntimeError.

Two GET routes are registered (mirroring OpenAPIExtension):

  • spec_path (default /asyncapi.json) → the document as JSON;
  • docs_path (default /asyncapi, None to skip) → an HTML page hosting the CDN AsyncAPI renderer.

init_app(app)

Wire the spec + docs routes onto app through the public API.

asyncapi_html(spec_url, title='BlackBull — AsyncAPI')

Return renderer HTML that loads its JavaScript from a CDN.

generate_asyncapi(mqtt, *, title, version, description, server_host)

Build an AsyncAPI 3.0 document for mqtt's registered on_message taps.

mqtt is an MQTTExtension (anything exposing iter_subscriptions() and port). The returned dict is JSON-serialisable.