# server faststream

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

Generate a Python [FastStream](https://faststream.ag2.ai) application of the Kafka, AMQP
(RabbitMQ) and MQTT channels of an AsyncAPI document.

## Usage

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

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

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

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

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

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