A bidirectional Python code generator that converts between AsyncAPI 3.0 specifications and Python code (pure Python or FastAPI implementations).
- 🔄 Bidirectional conversion between AsyncAPI 3.0 and Python code
- 🐍 Generate Python code from AsyncAPI 3.0 specifications:
- 🐍 Pure Python implementations with type hints
- ⚡ FastAPI endpoints with Pydantic models
- 📄 Generate AsyncAPI 3.0 specifications from Python code
- 🧠 Automatic type inference and mapping
- ⚡ Support for both async and sync functions
with uv:
uv tool install zen-generatorwith pipx:
pipx install zen-generatorwith uvx:
uvx zen-generatorImportant
Currently, only model and function definitions in the components block of the AsyncAPI file are supported. Inline definitions are not supported.
Note
This code snippet includes a custom definition for declaring required parameters in model/function definitions. Specifically, the required keyword is used to specify mandatory fields, as shown below:
required: - user_idThis ensures that the
user_idparameter is always provided when the model or function is utilized.
Convert between AsyncAPI 3.0 specifications and Python code:
# Generate FastAPI implementation from AsyncAPI spec uvx zen-generator fastapi # Generate pure Python implementation from AsyncAPI spec uvx zen-generator pure-python # Generate AsyncAPI spec from Python code uvx zen-generator asyncapi-documentationThe CLI is built with Typer and provides three main commands:
Usage:
$ [OPTIONS] COMMAND [ARGS]...Options:
--install-completion: Install completion for the current shell.--show-completion: Show completion for the current shell, to copy it or customize the installation.--help: Show this message and exit.
Commands:
asyncapi-documentationpure-pythonfastapi
Usage:
$ asyncapi-documentation [OPTIONS]Options:
--models-file PATH: [default: models.py]--functions-file PATH: [default: functions.py]--output-file PATH: [default: asyncapi.yaml]--application-name TEXT: [default: Zen]--help: Show this message and exit.
Usage:
$ pure-python [OPTIONS]Options:
--asyncapi-file PATH: [default: asyncapi.yaml]--models-file PATH: [default: models.py]--functions-file PATH: [default: functions.py]--application-name TEXT: [default: Zen]--is-async / --no-is-async: [default: no-is-async]--help: Show this message and exit.
Usage:
$ fastapi [OPTIONS]Options:
--asyncapi-file PATH: [default: asyncapi.yaml]--models-file PATH: [default: models.py]--functions-file PATH: [default: functions.py]--application-name TEXT: [default: Zen]--is-async / --no-is-async: [default: no-is-async]--help: Show this message and exit.
from __future__ importannotationsfromtypingimportTypedDictclassUserModel(TypedDict): id: intname: stremail: str|None=Nonefrom __future__ importannotationsfrom .modelsimportUserModeldefget_user(user_id: int) ->UserModel: ...from__future__annotationsfrompydanticimportBaseModelclassUserModel(BaseModel): id: intname: stremail: str|None=Nonefrom__future__annotationsfromfastapiimportFastAPIfrom .modelsimportUserModelapp=FastAPI() @app.get("/users/{user_id}")asyncdefget_user(user_id: int) ->UserModel: ...asyncapi: 3.0.0info: title: Zenversion: 0.0.1description: ''channels: get_user: $ref: '#/components/channels/get_user'operations: get_user: $ref: '#/components/operations/get_user'components: channels: get_user: messages: request: $ref: '#/components/messages/get_user_request'response: $ref: '#/components/messages/get_user_response'operations: get_user: action: receivedescription: ''channel: $ref: '#/channels/get_user'messages: - $ref: '#/channels/get_user/messages/request'reply: channel: $ref: '#/channels/get_user'messages: - $ref: '#/channels/get_user/messages/response'messages: get_user_request: title: Request params for get_usersummary: ''description: ''payload: type: objectrequired: - user_idproperties: user_id: type: integerdescription: ''get_user_response: title: Response params for get_usersummary: ''description: ''payload: $ref: '#/components/schemas/UserModel'format: requiredschemas: UserModel: type: objectbase_class: BaseModelrequired: - id - nameproperties: id: type: integername: type: stringemail: type: stringRequirements:
- Python 3.10+
- uv (Python packaging toolchain)
# Install uv if not already installed# see https://docs.astral.sh/uv/getting-started/installation/# Clone repository git clone https://github.com/WaYdotNET/zen-generator.git cd zen-generator # Install dependencies with uv uv sync # Run tests uv run pytestAsyncAPI Specification
- Follow AsyncAPI 3.0 guidelines
- Define clear schema types
- Include comprehensive examples
- Use semantic versioning
Code Generation
- Review generated code for correctness
- Implement business logic in function stubs
- Keep generated files synchronized
- Use type hints consistently
Project Organization
- Maintain clear separation between models and functions
- Follow standard Python package structure
- Implement proper error handling
Contributions are welcome! Please:
- Fork the repository
- Create a feature branch
- Commit your changes
- Push to the branch
- Submit a pull request
MIT License - see LICENSE file for details
- GitHub Issues: Report bugs or suggest features
Made with ❤️ by WaYdotNET
