Skip to content

get_app_schema

faststream.specification.asyncapi.v3_0_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/v3_0_0/generate.py 
 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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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]]] | None,
    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, operations = get_broker_channels(broker)

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

    for channel in channels.values():
        channel.servers = [
            {"$ref": f"#/servers/{server_name}"} for server_name in list(servers.keys())
        ]
    added_channels, added_operations = get_asgi_routes(http_handlers)
    channels.update(added_channels)
    operations.update(added_operations)

    for channel_name, channel in channels.items():
        msgs: dict[str, Message | Reference] = {}
        for message_name, message in channel.messages.items():
            assert isinstance(message, Message)

            msgs[message_name] = _resolve_msg_payloads(
                message_name,
                message,
                channel_name,
                payloads,
                messages,
            )

        channel.messages = msgs

    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 if tags else None,
            externalDocs=ExternalDocs.from_spec(external_docs),
        ),
        asyncapi=schema_version,
        defaultContentType=ContentTypes.JSON.value,
        id=identifier,
        servers=servers,
        channels=channels,
        operations=operations,
        components=Components(
            messages=messages,
            schemas=payloads,
            securitySchemes=None
            if broker.specification.security is None
            else broker.specification.security.get_schema(),
        ),
    )