Coverage for faststream / specification / asyncapi / site.py: 47%
30 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-08 01:48 +0000
1from functools import partial
2from http import server
3from typing import TYPE_CHECKING, Any
4from urllib.parse import parse_qs, urlparse
6from faststream._internal._compat import json_dumps
7from faststream._internal.logger import logger
9if TYPE_CHECKING:
10 from faststream.specification import Specification
12ASYNCAPI_JS_DEFAULT_URL = (
13 "https://unpkg.com/@asyncapi/react-component@3.0.2/browser/standalone/index.js"
14)
16ASYNCAPI_CSS_DEFAULT_URL = (
17 "https://unpkg.com/@asyncapi/react-component@3.0.2/styles/default.min.css"
18)
21ASYNCAPI_TRY_IT_PLUGIN_URL = "https://cdn.jsdelivr.net/npm/asyncapi-try-it-plugin@0.3.0-standalone.0/dist/index.iife.min.js"
24def get_asyncapi_html(
25 schema: "Specification",
26 sidebar: bool = True,
27 info: bool = True,
28 servers: bool = True,
29 operations: bool = True,
30 messages: bool = True,
31 schemas: bool = True,
32 errors: bool = True,
33 expand_message_examples: bool = True,
34 asyncapi_js_url: str = ASYNCAPI_JS_DEFAULT_URL,
35 asyncapi_css_url: str = ASYNCAPI_CSS_DEFAULT_URL,
36 try_it_out_plugin_url: str = ASYNCAPI_TRY_IT_PLUGIN_URL,
37 try_it_out: bool = True,
38 try_it_out_url: str = "asyncapi/try",
39) -> str:
40 """Generate HTML for displaying an AsyncAPI document."""
41 config = {
42 "show": {
43 "sidebar": sidebar,
44 "info": info,
45 "servers": servers,
46 "operations": operations,
47 "messages": messages,
48 "schemas": schemas,
49 "errors": errors,
50 },
51 "expand": {
52 "messageExamples": expand_message_examples,
53 },
54 "sidebar": {
55 "showServers": "byDefault",
56 "showOperations": "byDefault",
57 },
58 }
60 if try_it_out: 60 ↛ 87line 60 didn't jump to line 87 because the condition on line 60 was always true
61 plugins_js = f"""
62 <script src="{asyncapi_js_url}"></script>
63 <script src="{try_it_out_plugin_url}"></script>
64 <script>
65 const schema = {schema.to_json()};
66 const config = {json_dumps(config).decode()};
67 const endpoint = {try_it_out_url!r};
68 const plugin = window.AsyncApiTryItPlugin.createTryItOutPlugin({{
69 endpointBase: endpoint.replace(/^\\//, ""),
70 resolveEndpoint: () => endpoint,
71 showPayloadSchema: true,
72 showEndpointInput: false,
73 showRealBrokerToggle: true
74 }});
76 window.AsyncApiStandalone.render(
77 {{
78 schema,
79 config,
80 plugins: [plugin]
81 }},
82 document.getElementById("asyncapi")
83 );
84 </script>"""
86 else:
87 standalone_config = {"schema": schema.to_json(), "config": config}
88 plugins_js = f"""
89 <script src="{asyncapi_js_url}"></script>
90 <script>
91 window.AsyncApiStandalone.render(
92 {json_dumps(standalone_config).decode()},
93 document.getElementById("asyncapi")
94 );
95 </script>"""
97 return f"""<!DOCTYPE html>
98<html>
99 <head>
100 <title>{schema.title} AsyncAPI</title>
101 <link rel="icon" href="https://www.asyncapi.com/favicon.ico">
102 <link rel="icon" type="image/png" sizes="16x16" href="https://www.asyncapi.com/favicon-16x16.png">
103 <link rel="icon" type="image/png" sizes="32x32" href="https://www.asyncapi.com/favicon-32x32.png">
104 <link rel="icon" type="image/png" sizes="194x194" href="https://www.asyncapi.com/favicon-194x194.png">
105 <link rel="stylesheet" href="{asyncapi_css_url}">
106 </head>
107 <style>
108 html {{
109 font-family: ui-sans-serif,system-ui,-apple-system,BlinkMacSystemFont,Segoe UI,Roboto,Helvetica Neue,Arial,Noto Sans,sans-serif,Apple Color Emoji,Segoe UI Emoji,Segoe UI Symbol,Noto Color Emoji;
110 line-height: 1.5;
111 }}
112 </style>
113 <body>
114 <div id="asyncapi"></div>
115 {plugins_js}
116 </body>
117</html>"""
120def serve_app(
121 schema: "Specification",
122 host: str,
123 port: int,
124) -> None:
125 """Serve the HTTPServer with AsyncAPI schema."""
126 logger.info(f"HTTPServer running on http://{host}:{port} (Press CTRL+C to quit)")
127 logger.warning("Please, do not use it in production.")
129 server.HTTPServer(
130 (host, port),
131 partial(_Handler, schema=schema),
132 ).serve_forever()
135class _Handler(server.BaseHTTPRequestHandler):
136 def __init__(
137 self,
138 *args: Any,
139 schema: "Specification",
140 **kwargs: Any,
141 ) -> None:
142 self.schema = schema
143 super().__init__(*args, **kwargs)
145 def get_query_params(self) -> dict[str, bool]:
146 return {
147 i: _str_to_bool(next(iter(j))) if j else False
148 for i, j in parse_qs(urlparse(self.path).query).items()
149 }
151 def do_GET(self) -> None:
152 """Serve a GET request."""
153 query_dict = self.get_query_params()
155 encoding = "utf-8"
156 html = get_asyncapi_html(
157 self.schema,
158 sidebar=query_dict.get("sidebar", True),
159 info=query_dict.get("info", True),
160 servers=query_dict.get("servers", True),
161 operations=query_dict.get("operations", True),
162 messages=query_dict.get("messages", True),
163 schemas=query_dict.get("schemas", True),
164 errors=query_dict.get("errors", True),
165 expand_message_examples=query_dict.get("expandMessageExamples", True),
166 try_it_out=False, # CLI serve has no broker — use AsgiFastStream for try-it
167 )
168 body = html.encode(encoding=encoding)
170 self.send_response(200)
171 self.send_header("content-length", str(len(body)))
172 self.send_header("content-type", f"text/html; charset={encoding}")
173 self.end_headers()
174 self.wfile.write(body)
177def _str_to_bool(v: str) -> bool:
178 return v.lower() in {"1", "t", "true", "y", "yes"}