AsyncAPI Documentation

generate_asyncapi_doc(routes: list[RouteDefinition], config: AsyncAPIGeneratorConfig | None = None) -> dict[str, Any]

Generate an AsyncAPI 2.6.0 document from route definitions.

Parameters:

Name	Type	Description	Default
routes	list[RouteDefinition]	List of RouteDefinition from broker.routes.	required
config	AsyncAPIGeneratorConfig | None	Generator configuration.	None

Returns:

Type	Description
dict[str, Any]	AsyncAPI 2.6.0 spec as a JSON-serializable dict.

Source code in src/rabbitkit/asyncapi/generator.py

def generate_asyncapi_doc(
    routes: list[RouteDefinition],
    config: AsyncAPIGeneratorConfig | None = None,
) -> dict[str, Any]:
    """Generate an AsyncAPI 2.6.0 document from route definitions.

    Args:
        routes: List of RouteDefinition from broker.routes.
        config: Generator configuration.

    Returns:
        AsyncAPI 2.6.0 spec as a JSON-serializable dict.
    """
    if config is None:
        config = AsyncAPIGeneratorConfig()

    doc: dict[str, Any] = {
        "asyncapi": "2.6.0",
        "info": {
            "title": config.title,
            "version": config.version,
        },
        "servers": {
            "rabbitmq": {
                "url": config.server_url,
                "protocol": "amqp",
                "description": config.server_description,
            },
        },
        "channels": {},
    }

    if config.description:
        doc["info"]["description"] = config.description

    for route in routes:
        channel_name = route.queue.name
        channel: dict[str, Any] = {}

        if route.description:
            channel["description"] = route.description

        # Subscribe operation (consumer)
        operation: dict[str, Any] = {
            "operationId": route.name,
        }

        # Message schema from handler type hints
        body_type = get_handler_body_type(route.handler)
        payload = extract_json_schema(body_type)
        message: dict[str, Any] = {
            "name": route.name,
        }
        if payload:
            message["payload"] = payload
        operation["message"] = message

        # Tags
        if route.tags:
            operation["tags"] = [{"name": t} for t in sorted(route.tags)]

        channel["subscribe"] = operation

        # AMQP bindings
        bindings: dict[str, Any] = {"amqp": {"is": "queue"}}
        queue_binding: dict[str, Any] = {
            "name": route.queue.name,
            "durable": route.queue.durable,
        }
        if hasattr(route.queue, "exclusive"):
            queue_binding["exclusive"] = route.queue.exclusive
        bindings["amqp"]["queue"] = queue_binding

        if route.exchange is not None:
            exchange_binding: dict[str, Any] = {
                "name": route.exchange.name,
                "type": (
                    route.exchange.type.value
                    if hasattr(route.exchange.type, "value")
                    else str(route.exchange.type)
                ),
            }
            if hasattr(route.exchange, "durable"):
                exchange_binding["durable"] = route.exchange.durable
            bindings["amqp"]["exchange"] = exchange_binding

        channel["bindings"] = bindings

        # Publish operation (if result_publisher is set)
        if route.result_publisher is not None:
            publish_op: dict[str, Any] = {
                "operationId": f"{route.name}.reply",
                "message": {"name": f"{route.name}.response"},
            }
            channel["publish"] = publish_op

        doc["channels"][channel_name] = channel

    return doc

generate_asyncapi_json(routes: list[RouteDefinition], config: AsyncAPIGeneratorConfig | None = None, indent: int | None = 2) -> str

Generate AsyncAPI spec as a JSON string.

Source code in src/rabbitkit/asyncapi/generator.py

def generate_asyncapi_json(
    routes: list[RouteDefinition],
    config: AsyncAPIGeneratorConfig | None = None,
    indent: int | None = 2,
) -> str:
    """Generate AsyncAPI spec as a JSON string."""
    doc = generate_asyncapi_doc(routes, config)
    return json.dumps(doc, indent=indent)

AsyncAPIGeneratorConfig dataclass

Configuration for AsyncAPI document generation.

Source code in src/rabbitkit/asyncapi/generator.py

@dataclass
class AsyncAPIGeneratorConfig:
    """Configuration for AsyncAPI document generation."""

    title: str = "rabbitkit Service"
    version: str = "1.0.0"
    description: str = ""
    server_url: str = "localhost:5672"
    server_description: str = "RabbitMQ"