API Genie

Generate a Python FastStream application of the Kafka, AMQP (RabbitMQ) and MQTT channels of an AsyncAPI document.

Usage

apigenie-generator server faststream [...OPTIONS] INPUT_FILE

Description

The input must be an AsyncAPI 3.x document. Documents referencing other files are bundled first.

The application is the one the document describes: it receives the messages of receive operations and sends those of send operations.

one broker per server: KafkaBroker for kafka and kafka-secure servers (aiokafka), RabbitBroker for amqp and amqps (AMQP 0.9.1) and MQTTBroker for mqtt, mqtts and secure-mqtt (MQTT 5 or 3.1.1). A broker connects to the URL in an environment variable named after its server (PRODUCTION_KAFKA_URL), or to the server’s address with the default values of its variables
a subscriber per receive operation, @broker.subscriber(...), which calls a handler you write
a publisher per send operation, @broker.publisher(...), and an async function publishing through it with typed arguments
Pydantic v2 models of the payloads and headers

from warehouse_events import models
from warehouse_events.publishers import publish_audit, request_quote

await publish_audit(models.AuditEntry(action="login"), region="eu")
quote = await request_quote(models.QuoteRequest(sku="B"), timeout=5)

Run the application with faststream run <package>.app:app.

Channels and bindings

Document	Generated code
Channel address, Kafka binding `topic`	the topic; with parameters a `pattern=` subscription and `Path()` parameters
Kafka operation binding `groupId` (`const` or a single `enum` value), also from operation traits	`group_id=` of the subscriber
AMQP binding `is: queue`	a `RabbitQueue` (name, `durable`, `exclusive`, `auto_delete`) the operations publish to and consume from
AMQP binding `is: routingKey` with an exchange	a `RabbitExchange` in `brokers.py`; the address is the routing key, with parameters a topic pattern and `Path()` parameters. Subscribers bind a queue named after the operation
AMQP `vhost`	`virtualhost=` of the broker
AMQP operation binding `priority`, `expiration`, `deliveryMode: 2`, `mandatory`, `userId`, `timestamp`	the publisher’s `priority`, `expiration` (in seconds), `persist`, `mandatory`, `user_id`; a timestamp per message
MQTT topic parameters (whole topic levels)	`Path()` parameters
MQTT operation binding `qos`, `retain`	`qos=`, `retain=`
MQTT server binding `clientId`, `cleanSession`, `keepAlive`, `sessionExpiryInterval`, `lastWill`	the broker’s `client_id`, `clean_session`, `keepalive`, `session_expiry_interval`, `will`

Messages

A JSON payload becomes a Pydantic model, a text/* one str and any other content type bytes. A payload referencing a schema of components/schemas is the model of that schema; other payloads and headers get a model named after their message (OrderCreated, OrderCreatedHeaders). The headers of message traits are merged into the message’s own as AsyncAPI specifies (JSON Merge Patch: the message over its traits, later traits over earlier ones).

Subscribers take the headers as typed Header() parameters and pass them to the handlers as keyword arguments; publish functions take a headers model. When several messages of a channel can be told apart by a property with a distinct constant value (type: { const: created }), the payload is a discriminated union and every message has a handler of its own (consume_orders_order_created); otherwise the handler of the operation gets the union.

Request/reply

A send operation with a reply channel: its function publishes the request and returns the reply, waiting at most timeout seconds. A subscriber of the reply channel hands replies to the waiting requests by correlation id: the message’s correlationId (a header, which the function fills in when the request has none, or a payload property), or FastStream’s correlation id when the document declares none. FastStream’s broker.request() isn’t used: Kafka doesn’t support it outside the test client, and RabbitMQ’s waits on amq.rabbitmq.reply-to instead of the reply channel of the document.
A send operation with a reply address in a header ($message.header#/replyTo): its function takes a reply_to argument and sends it in that header.
A receive operation with a reply channel: the handler returns the reply, which is published to the reply channel with the request’s correlation id header.
A receive operation with a reply address in a header: the handler returns the reply, which is published to the address in the request’s header.

Output

pyproject.toml          faststream[cli,…], pydantic; `pip install -e .`
README.md               how to run, the environment variables, what isn't generated
<package>/app.py        the FastStream application
<package>/brokers.py    the brokers and AMQP exchanges
<package>/subscribers.py
<package>/publishers.py
<package>/handlers.py   the handlers you write
<package>/models.py     Pydantic models (shared with apigenie-generator client python)
<package>/_models.py, _runtime.py, __init__.py, py.typed

handlers.py is written only when it doesn’t exist: it holds your code. After regenerating, add the handlers of new operations to it; mypy reports the missing ones.

Not generated

Reported as info lines or warnings:

channels of WebSocket and HTTP servers, which FastStream has no broker for
messages with Avro (or other non-JSON Schema) payloads, and operations left without messages
security schemes: give the brokers a security object (faststream.security) in brokers.py
servers no generated operation uses; a channel with several servers is served by its first one (choose others with the servers option)
reply addresses in payloads, AMQP cc and bcc, MQTT topic parameters inside a topic level

Extensions

Extension	Effect
`x-apigenie.codegen.name`	on a component schema: the name of its model class; on an operation: the name of its function (in snake_case)
`x-enum-varnames`, `x-enumNames`	names of the members of an enum
`x-apigenie.codegen.disabled`	`true` on a channel or an operation leaves it out (with the operations of a channel left out)

The extensions can live in an overlay for this target instead of in the document, e.g. faststream.overlay.yaml:

overlay: 1.0.0
info:
  title: FastStream names
  version: 1.0.0
actions:
  - target: $.operations.publishOrder
    update:
      x-apigenie:
        codegen:
          name: announce_order
  - target: $.components.schemas.OrderStatus
    update:
      x-enum-varnames: [CREATED, PAID, SHIPPED]

apigenie-generator server faststream asyncapi.yaml --overlay faststream.overlay.yaml -o out

Options

-c, –config

Path to the configuration file to use. See the “Configuration” section for more information.

–overlay OVERLAY

Path to an OpenAPI Overlay (1.0 to 1.2) applied to the input before generating, e.g. one setting the extensions of the “Extensions” section. Can be repeated: the overlays apply in order, after those listed under commands.generate.server_faststream.overlay in the configuration file (with paths relative to it). Warnings and errors about a value an overlay set point to the overlay. The overlays apply to the AsyncAPI document bundled with the files it references.

–strict

Makes an overlay action whose target selects nothing an error (exit status 2, nothing is generated) instead of a warning (OVERLAY:LINE:COL: action #N (TARGET) selects no nodes.).

-o, –output

Directory where the package is generated.

–package-name

Name of the generated distribution; the Python package is its snake_case form. Defaults to the title of the API.

–package-version

Version of the generated package. Defaults to the version of the API.

Configuration

version: '1.0'
commands:
  generate:
    server_faststream:
      servers: [production-kafka, customs-broker]
      enum_style: enum
      field_prefix: field_
      class_overrides:
        ShipmentId: ShipmentKey

Option	Description
`servers`	Keys of the servers to generate brokers for; all Kafka, AMQP and MQTT servers by default
`enum_style`	`enum` (default): `class Status(str, Enum)` for enums of component schemas; `literal`: `Literal[...]` everywhere
`field_prefix`	Prefix of field names that aren’t valid identifiers or clash with Pydantic, e.g. `field_json` (default `field_`)
`class_overrides`	Class names of component schemas, by schema name