AsyncFast¶
AsyncFast is a modern, event framework for building APIs with Python based on standard Python type hints.
Key Features:
Portable: Following AMGI should allow for implementations of any protocol
Standards-based: Based on AsyncAPI and JSON Schema
Terminology¶
Terminology throughout this documentation will follow AsyncAPI where possible, this means using terms like channel
instead of topic etc.
AMGI¶
AMGI (Asynchronous Messaging Gateway Interface) is the spiritual sibling of ASGI. While the focus of ASGI is HTTP, the focus here is event-based applications.
Warning
Given this project is in the very early stages treat AMGI as unstable
Installation¶
pip install "asyncfast[standard]"
Basic Usage¶
The simplest AsyncFast could be:
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"
}
}
}
}
Running¶
To run the app install an AMGI server (at the moment there is only amgi-aiokafka) then run:
$ asyncfast run amgi-aiokafka main:app channel
AsyncAPI Generation¶
$ asyncfast asyncapi main:app
Requirements¶
This project stands on the shoulders of giants:
Pydantic for the data parts.
Taking ideas from: