# client python (AsyncAPI)

Source: https://apigenie.pl/techdocs/reference/cli-generate_client_python_asyncapi

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

## Usage

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

## Description

`apigenie-generator client python` generates from OpenAPI documents (see
[apigenie-generator client python](/techdocs/reference/cli-generate_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](https://github.com/aio-libs/aiokafka), with [Pydantic](https://docs.pydantic.dev) 2
models of the payloads: the models of
[apigenie-generator server faststream](/techdocs/reference/cli-generate_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:

```python
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:

```python
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`:

```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]
```

```sh
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.

```yaml
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](/techdocs/reference/cli-generate_client_python#configuration) |