Automatic Specification module
Creating automatic specification
# specification.py
from user_events import receive_user_update
from asyncapi import AutoSpec
spec = AutoSpec('User Events', development='redis://localhost')
spec.subscribe(receive_user_update, channel_name='user/update')
Creating subscribers module
# user_events.py
import dataclasses
from typing import Optional
@dataclasses.dataclass
class UserUpdateMessage:
id: str
name: Optional[str] = None
age: Optional[int] = None
async def receive_user_update(message: UserUpdateMessage) -> None:
print(f"Received update for user id={message.id}")
Start subscriber to listen events
PYTHONPATH=. asyncapi-subscriber --api-module specification
Publishing Updates
# publish.py
import asyncio
from asyncapi import build_api_auto_spec
api = build_api_auto_spec('specification')
channel_id = 'user/update'
message = api.payload(channel_id, id='fake-user', name='Fake User', age=33)
async def publish() -> None:
await api.connect()
await api.publish(channel_id, message)
await api.disconnect()
asyncio.run(publish())
print(f"Published update for user={message.id}")
python publish.py
Published update for user=fake-user
Receive Updates
Waiting messages...
Received update for user id=fake-user
Expose Specification
PYTHONPATH=. asyncapi-docs --api-module specification
curl -i localhost:5000/asyncapi.yaml