Message Payload¶
In a decorated channel any argument that is not a channel parameter, or header is treated as the payload.
Important
Only one payload argument is allowed, there is no attempt to merge multiple, instead an error will be raised
BaseModel¶
You can declare a data model as a class that inherits from BaseModel. This can then be used as the payload argument:
from asyncfast import AsyncFast
from pydantic import BaseModel
app = AsyncFast()
class Payload(BaseModel):
id: str
name: str
@app.channel("channel")
async def handle_channel(payload: Payload) -> None:
print(payload)
{
"asyncapi": "3.0.0",
"info": {
"title": "AsyncFast",
"version": "0.1.0"
},
"channels": {
"HandleChannel": {
"address": "channel",
"messages": {
"HandleChannelMessage": {
"$ref": "#/components/messages/HandleChannelMessage"
}
}
}
},
"operations": {
"receiveHandleChannel": {
"action": "receive",
"channel": {
"$ref": "#/channels/HandleChannel"
}
}
},
"components": {
"messages": {
"HandleChannelMessage": {
"payload": {
"$ref": "#/components/schemas/Payload"
}
}
},
"schemas": {
"Payload": {
"properties": {
"id": {
"title": "Id",
"type": "string"
},
"name": {
"title": "Name",
"type": "string"
}
},
"required": [
"id",
"name"
],
"title": "Payload",
"type": "object"
}
}
}
}
List¶
from asyncfast import AsyncFast
from pydantic import BaseModel
app = AsyncFast()
class Item(BaseModel):
id: str
name: str
@app.channel("channel")
async def handle_channel(items: list[Item]) -> None:
print(items)
{
"asyncapi": "3.0.0",
"info": {
"title": "AsyncFast",
"version": "0.1.0"
},
"channels": {
"HandleChannel": {
"address": "channel",
"messages": {
"HandleChannelMessage": {
"$ref": "#/components/messages/HandleChannelMessage"
}
}
}
},
"operations": {
"receiveHandleChannel": {
"action": "receive",
"channel": {
"$ref": "#/channels/HandleChannel"
}
}
},
"components": {
"messages": {
"HandleChannelMessage": {
"payload": {
"type": "array",
"items": {
"$ref": "#/components/schemas/Item"
}
}
}
},
"schemas": {
"Item": {
"properties": {
"id": {
"title": "Id",
"type": "string"
},
"name": {
"title": "Name",
"type": "string"
}
},
"required": [
"id",
"name"
],
"title": "Item",
"type": "object"
}
}
}
}
Dataclass¶
from dataclasses import dataclass
from asyncfast import AsyncFast
app = AsyncFast()
@dataclass
class Order:
id: str
skus: list[str]
@app.channel("order")
async def handle_order(order: Order) -> None:
print(order)
{
"asyncapi": "3.0.0",
"info": {
"title": "AsyncFast",
"version": "0.1.0"
},
"channels": {
"HandleOrder": {
"address": "order",
"messages": {
"HandleOrderMessage": {
"$ref": "#/components/messages/HandleOrderMessage"
}
}
}
},
"operations": {
"receiveHandleOrder": {
"action": "receive",
"channel": {
"$ref": "#/channels/HandleOrder"
}
}
},
"components": {
"messages": {
"HandleOrderMessage": {
"payload": {
"$ref": "#/components/schemas/Order"
}
}
},
"schemas": {
"Order": {
"properties": {
"id": {
"title": "Id",
"type": "string"
},
"skus": {
"items": {
"type": "string"
},
"title": "Skus",
"type": "array"
}
},
"required": [
"id",
"skus"
],
"title": "Order",
"type": "object"
}
}
}
}
Built in¶
from asyncfast import AsyncFast
app = AsyncFast()
@app.channel("channel")
async def handle_channel(payload: int) -> None:
print(payload)
{
"asyncapi": "3.0.0",
"info": {
"title": "AsyncFast",
"version": "0.1.0"
},
"channels": {
"HandleChannel": {
"address": "channel",
"messages": {
"HandleChannelMessage": {
"$ref": "#/components/messages/HandleChannelMessage"
}
}
}
},
"operations": {
"receiveHandleChannel": {
"action": "receive",
"channel": {
"$ref": "#/channels/HandleChannel"
}
}
},
"components": {
"messages": {
"HandleChannelMessage": {
"payload": {
"type": "integer"
}
}
}
}
}
Avro¶
By default a payload is parsed as JSON. Annotating it with AvroPayload parses it as Avro binary instead, from a
schema derived from the same type hints:
from typing import Annotated
from asyncfast import AsyncFast
from asyncfast_avro import AvroPayload
from pydantic import BaseModel
app = AsyncFast()
class Order(BaseModel):
id: str
skus: list[str]
@app.channel("order")
async def handle_order(order: Annotated[Order, AvroPayload()]) -> None:
print(order)
{
"asyncapi": "3.0.0",
"info": {
"title": "AsyncFast",
"version": "0.1.0"
},
"channels": {
"HandleOrder": {
"address": "order",
"messages": {
"HandleOrderMessage": {
"$ref": "#/components/messages/HandleOrderMessage"
}
}
}
},
"operations": {
"receiveHandleOrder": {
"action": "receive",
"channel": {
"$ref": "#/channels/HandleOrder"
}
}
},
"components": {
"messages": {
"HandleOrderMessage": {
"payload": {
"schemaFormat": "application/vnd.apache.avro;version=1.9.0",
"schema": {
"type": "record",
"name": "Order",
"fields": [
{
"name": "id",
"type": "string"
},
{
"name": "skus",
"type": {
"type": "array",
"items": "string"
}
}
]
}
},
"contentType": "application/vnd.apache.avro;version=1.9.0"
}
}
}
}
The schema is published in the generated AsyncAPI document as a Multi Format Schema Object, so consumers in other languages can be generated from it.
Avro support is provided by the separate asyncfast-avro package:
pip install asyncfast-avro==0.47.0
Important
Payloads are encoded as plain Avro binary, not the Confluent wire format, so no schema registry is involved. The schema in the AsyncAPI document is both the writer and the reader schema: a single self-contained schema per payload, with no schema evolution and no reader/writer distinction.
Type mapping¶
Avro schemas are derived from the JSON schema pydantic generates, so the types below can be used as Avro payloads. Types
without a direct Avro equivalent are carried as strings. Types Avro cannot represent, such as object, untyped
mappings and collections, or unnamed records, are rejected when the schema is generated:
Python |
Avro |
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Note
Naive datetime values (with no timezone) are encoded as UTC. Field defaults are emitted in the schema, except
for fields of nested models, where a default would require expanding the defaults of all of the nested fields.