Skip to content

get_app_schema

faststream.specification.asyncapi.v2_6_0.generate.get_app_schema #

get_app_schema(
    broker,
    /,
    title,
    app_version,
    schema_version,
    description,
    terms_of_service,
    contact,
    license,
    identifier,
    tags,
    external_docs,
    http_handlers,
)

Get the application schema.

Source code in faststream/specification/asyncapi/v2_6_0/generate.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get_app_schema(
    broker: "BrokerUsecase[Any, Any]",
    /,
    title: str,
    app_version: str,
    schema_version: str,
    description: str | None,
    terms_of_service: Optional["AnyHttpUrl"],
    contact: Union["SpecContact", "ContactDict", dict[str, Any]] | None,
    license: Union["SpecLicense", "LicenseDict", dict[str, Any]] | None,
    identifier: str | None,
    tags: Sequence[Union["SpecTag", "TagDict", dict[str, Any]]],
    external_docs: Union["SpecDocs", "ExternalDocsDict", dict[str, Any]] | None,
    http_handlers: list[tuple[str, "HttpHandler"]],
) -> ApplicationSchema:
    """Get the application schema."""
    servers = get_broker_server(broker)
    channels = get_broker_channels(broker)

    messages: dict[str, Message] = {}
    payloads: dict[str, dict[str, Any]] = {}

    for channel in channels.values():
        channel.servers = list(servers.keys())
    channels.update(get_asgi_routes(http_handlers))

    for channel_name, ch in channels.items():
        resolve_channel_messages(ch, channel_name, payloads, messages)

    return ApplicationSchema(
        info=ApplicationInfo(
            title=title,
            version=app_version,
            description=description,
            termsOfService=terms_of_service,
            contact=Contact.from_spec(contact),
            license=License.from_spec(license),
        ),
        tags=[Tag.from_spec(tag) for tag in tags] or None,
        externalDocs=ExternalDocs.from_spec(external_docs),
        asyncapi=schema_version,
        defaultContentType=ContentTypes.JSON.value,
        id=identifier,
        servers=servers,
        channels=channels,
        components=Components(
            messages=messages,
            schemas=payloads,
            securitySchemes=None
            if broker.specification.security is None
            else broker.specification.security.get_schema(),
        ),
    )