Skip to content

make_asyncapi_asgi

faststream.asgi.make_asyncapi_asgi #

make_asyncapi_asgi(
    app,
    sidebar=True,
    info=True,
    servers=True,
    operations=True,
    messages=True,
    schemas=True,
    errors=True,
    expand_message_examples=True,
    title="FastStream",
    asyncapi_js_url=ASYNCAPI_JS_DEFAULT_URL,
    asyncapi_css_url=ASYNCAPI_CSS_DEFAULT_URL,
)
Source code in faststream/asgi/factories.py 
40
41
42
43
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
def make_asyncapi_asgi(
    app: "AsyncAPIApplication",
    sidebar: bool = True,
    info: bool = True,
    servers: bool = True,
    operations: bool = True,
    messages: bool = True,
    schemas: bool = True,
    errors: bool = True,
    expand_message_examples: bool = True,
    title: str = "FastStream",
    asyncapi_js_url: str = ASYNCAPI_JS_DEFAULT_URL,
    asyncapi_css_url: str = ASYNCAPI_CSS_DEFAULT_URL,
) -> "ASGIApp":
    cached_docs = None

    @get
    async def docs(scope: "Scope") -> AsgiResponse:
        nonlocal cached_docs
        if not cached_docs:
            cached_docs = get_asyncapi_html(
                get_app_schema(app),
                sidebar=sidebar,
                info=info,
                servers=servers,
                operations=operations,
                messages=messages,
                schemas=schemas,
                errors=errors,
                expand_message_examples=expand_message_examples,
                title=title,
                asyncapi_js_url=asyncapi_js_url,
                asyncapi_css_url=asyncapi_css_url,
            )
        return AsgiResponse(
            cached_docs.encode("utf-8"),
            200,
            {"Content-Type": "text/html; charset=utf-8"},
        )

    return docs