client python (AsyncAPI)

Generate a Python asyncio Kafka client package for the Kafka channels of an AsyncAPI document.

Usage

apigenie-generator client python [...OPTIONS] INPUT_FILE

Description

apigenie-generator client python generates from OpenAPI documents (see apigenie-generator client python) and from AsyncAPI 3.x documents, telling them apart by their openapi or asyncapi field. Documents referencing other files are bundled first.

From an AsyncAPI document it generates a Python package (Python 3.10 or later) on aiokafka, with Pydantic 2 models of the payloads: the models of apigenie-generator server faststream. By default the client is the counterpart of the application the document describes: it sends the messages of the application’s receive operations and receives those of its send operations. Set the perspective option to application to generate the described application’s side instead.

The generator writes the Kafka channels: the channels of servers with the kafka or kafka-secure protocol, and channels with a Kafka binding in documents without servers. Channels of other protocols are skipped and reported as info.

With perspective: application, the client of the Carpet Express Event Platform sends the shipment events and requests labels:

options = CarpetExpressEventsKafkaClientOptions(bootstrap_servers="localhost:9092")
async with CarpetExpressEventsKafkaClient(options) as client:
    await client.send_shipment_delivered(delivered)

    # request/reply: returns the reply carrying the request's correlation id
    reply = await client.request_label_render(request, timeout=10.0)
    if isinstance(reply, messages.LabelRendered):
        print(reply.payload.url)

In the default perspective, it receives them and renders the labels:

class LabelFarm(RequestLabelRenderHandler):
    async def on_render_label(self, message: messages.RenderLabel) -> messages.RequestLabelRenderReply | None:
        # sent to the labelRenderResults topic with the request's correlation id
        return messages.LabelRendered(payload=render(message.payload), key=message.key)

subscription = await client.listen_request_label_render(LabelFarm())
await subscription.assigned()  # once the consumer group assigned partitions

The package is the snake case of --package-name, or of the document’s title followed by _client; the distribution name is its kebab case:

File	Contents
`models.py`, `_models.py`	the Pydantic models of the payloads
`messages.py`	a dataclass per message (`payload`, `headers`, `key`), its headers model with the constant headers (`CE_TYPE`, …), and a type alias per reply union
`client.py`	`<Name>KafkaClient`, `<Name>KafkaClientOptions` and a handler class per received operation
`_kafka.py`, `_runtime.py`	the Kafka runtime (records, subscriptions, replies) and the support of headers and correlation ids
`pyproject.toml`	the distribution, requiring aiokafka and Pydantic

The client. It is an async context manager; start() starts a producer per sending and replying operation, with the operation’s client id, and close() stops the subscriptions, the reply consumer and the producers. The options default to the document’s server, topics, client ids and group ids. They also take the reply client id and timeout (30 seconds), and keyword arguments of the aiokafka producers and consumers that replace the client’s (security settings, for example).
Sent operations. send_<message>(message) writes the message’s key, its headers (constant headers such as the ce_specversion and ce_type of CloudEvents included) and its JSON payload to the operation’s topic, and returns once the broker acknowledged it.
Request/reply with a reply channel: request_<operation>(message, *, timeout=None) sets a new correlation id unless the message has one (in its header, at its payload pointer, or in kafka_correlationId), sends the request and returns the reply, or raises TimeoutError. With several reply messages, the reply is one of them. One consumer, without a consumer group, receives the replies of all requests and matches them by correlation id. An operation named after the verb isn’t repeated: requestLabelRender gives request_label_render.
Received operations. listen_<operation>(handler) consumes the operation’s topic in a task, in its consumer group (new groups start at the first record), until the returned subscription or the client is closed. It commits the offsets of the records it handled.
- The handler class (PublishShipmentEventsHandler) has an abstract on_<message> method per message, an on_unknown and an on_error.
- Several messages are told apart by a header or payload property with a distinct constant value (e.g. ce_type). Otherwise the first message whose payload validates without unknown properties is chosen.
- Records that aren’t valid messages, and exceptions of the handler, go to on_error.
- A reply returned by the handler is sent to the reply topic with the request’s correlation id.

Messages are validated by their models when they are built and when they are received; headers models ignore the record’s other headers. The package passes mypy --strict.

Not supported, and reported as warnings: Avro payloads and payloads of other non-JSON schema formats (the messages are skipped, and operations left without a message too), reply addresses given by runtime expressions, and Kafka topics with parameters. Security (SASL/SCRAM, mTLS) is not configured: pass security_protocol, sasl_* and ssl_context in producer_options and consumer_options.

Extensions

Extension	Effect
`x-apigenie.codegen.name`	name of the model class of a component schema
`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. python.overlay.yaml:

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

apigenie-generator client python asyncapi.yaml --overlay python.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.client_python.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 distribution, by default the document’s title followed by -client. Its kebab case is the base of default group and client ids, and its Pascal case the base of the class names.

–package-version

Version of the distribution; defaults to the version of the API.

Configuration

The options are read from the asyncapi section of the configuration of apigenie-generator client python; its other keys configure the generation from OpenAPI documents.

version: '1.0'
commands:
  generate:
    client_python:
      asyncapi:
        perspective: counterpart
        enum_style: literal

Option	Description
`perspective`	`counterpart` (default): the counterpart of the described application; `application`: the described application, sending what it sends
`enum_style`, `field_prefix`, `class_overrides`	Settings of the Pydantic models, as for apigenie-generator client python