Skip to content

get_app_schema

faststream.specification.asyncapi.v2_6_0.get_app_schema #

get_app_schema(
    *brokers: BrokerUsecase[Any, Any, Any],
    title: str,
    app_version: str,
    schema_version: str,
    description: str | None,
    terms_of_service: Optional[AnyHttpUrl],
    contact: Union[Contact, ContactDict, dict[str, Any]]
    | None,
    license: Union[License, LicenseDict, dict[str, Any]]
    | None,
    identifier: str | None,
    tags: Sequence[Union[Tag, TagDict, dict[str, Any]]],
    external_docs: Union[
        ExternalDocs, ExternalDocsDict, dict[str, Any]
    ]
    | None,
    http_handlers: list[tuple[str, HttpHandler]],
) -> ApplicationSchema

Get the application schema.

Source code in faststream/specification/asyncapi/v2_6_0/generate.py 
 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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def get_app_schema(
    *brokers: "BrokerUsecase[Any, 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."""
    if any(br.specification.security for br in brokers):
        list_of_specification_security = (
            br.specification.security.get_schema()
            for br in brokers
            if br.specification.security
        )
        security_schemes = convert_list_of_dict_to_dict(
            list_of_specification_security,
            "specification security",
        )
    else:
        security_schemes = None

    servers, broker_servers = get_broker_server(*brokers)

    channels = convert_list_of_dict_to_dict(
        (get_broker_channels(br, servers=srv) for br, srv in broker_servers.items()),
        "channel",
    )
    channels.update(get_asgi_routes(http_handlers))

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

    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=security_schemes,
        ),
    )