AsyncAPI (v2) specification schema as pydantic class
Project description
asyncapi-schema-pydantic
AsyncAPI (v2) specification schema as Pydantic classes.
The naming of the classes follows the schema in AsyncAPI specification.
Installation
pip install asyncapi-schema-pydantic
Try me
from asyncapi_schema_pydantic import ( AsyncAPI,
Info,
ChannelItem,
Operation,
Message,
ChannelBindings,
AmqpChannelBinding,
AmqpQueue,
Tag )
# Construct AsyncAPI by pydantic objects
async_api = AsyncAPI(
info=Info(
title="Email Service",
version="1.0.0",
description='description'
),
channels={
"user/signedup": ChannelItem(
description='This channel is used to exchange messages about users signing up',
subscribe=Operation(
summary='A user signed up.',
message=Message(
name='UserSignup',
title='User signup',
summary='Action to sign a user up.',
description='A longer description of the message',
contentType='application/json',
tags=[
Tag(name='user'),
Tag(name='signup'),
Tag(name='register')
]
),
),
bindings=ChannelBindings(
amqp=AmqpChannelBinding(
param_is='queue',
queue=AmqpQueue(
name='my-queue-name',
durable=True,
exclusive=True,
autoDelete=False,
vhost='/'
)
)
)
)
}
)
print(async_api.json(by_alias=True, exclude_none=True, indent=2))
Result:
{
"asyncapi": "2.3.0",
"info": {
"title": "Email Service",
"version": "1.0.0",
"description": "description"
},
"channels": {
"user/signedup": {
"description": "This channel is used to exchange messages about users signing up",
"subscribe": {
"summary": "A user signed up.",
"message": {
"contentType": "application/json",
"name": "UserSignup",
"title": "User signup",
"summary": "Action to sign a user up.",
"description": "A longer description of the message",
"tags": [
{
"name": "user"
},
{
"name": "signup"
},
{
"name": "register"
}
]
}
},
"bindings": {
"amqp": {
"queue": {
"name": "my-queue-name",
"durable": true,
"exclusive": true,
"autoDelete": false,
"vhost": "/"
}
}
}
}
}
}
Take advantage of Pydantic
Pydantic is a great tool, allow you to use object / dict / mixed data for for input.
The following examples give the same AsyncAPI result as above:
from asyncapi_schema_pydantic import AsyncAPI, ChannelItem, Operation
# Construct AsyncAPI from dict
async_api = AsyncAPI.parse_obj({
"asyncapi": "2.3.0",
"info": {
"title": "Email Service",
"version": "1.0.0",
"description": "description"
},
"channels": {
"user/signedup": {
"description": "This channel is used to exchange messages about users signing up",
"subscribe": {
"summary": "A user signed up.",
"message": {
"contentType": "application/json",
"name": "UserSignup",
"title": "User signup",
"summary": "Action to sign a user up.",
"description": "A longer description of the message"
}
}
}
}
})
# Construct AsyncAPI with mix of dict/object
async_api = AsyncAPI.parse_obj({
"asyncapi": "2.3.0",
"info": {
"title": "Email Service",
"version": "1.0.0",
"description": "description"
},
"channels": {
"user/signedup": ChannelItem(
description='This channel is used to exchange messages about users signing up',
subscribe=Operation(
summary='A user signed up.',
message={
"contentType": "application/json",
"name": "UserSignup",
"title": "User signup",
"summary": "Action to sign a user up.",
"description": "A longer description of the message"
}
)
)
}
})
Load and validate an AsyncAPI specification from a YAML file
from asyncapi_schema_pydantic import AsyncAPI
async_api = AsyncAPI.load_from_file("tests/data/sample.yaml")
print(async_api.json(by_alias=True, exclude_none=True, indent=2))
License
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Close
Hashes for asyncapi-schema-pydantic-1.0.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | 97ee315249df15ea9c6befc24cdd69e29055ed9c2e5633003b62051005bc97ec |
|
MD5 | 92f27d88528dbaf4ed70e25cddfdca77 |
|
BLAKE2b-256 | 628704af445afd9e263d76fde295a4cdb75c70a51b9608c7ef22a162c3459709 |
Close
Hashes for asyncapi_schema_pydantic-1.0.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | e811599567ffaeced794cfea6b0a4af283165c09f460f6b45f85318ed2c6cd0e |
|
MD5 | 92ecdfe07313f6aaceff614cebe05e55 |
|
BLAKE2b-256 | 350e83b0b1124af899145b8fd86dc258217fede4b91587b4191e9226f2c0f231 |