Coverage for faststream / asgi / factories / asyncapi / docs.py: 90%

1from collections.abc import Sequence (empty)

2from typing import TYPE_CHECKING, Any, Union (empty)

3 

4from faststream.asgi.handlers import GetHandler, get (empty)

5from faststream.asgi.response import AsgiResponse (empty)

6from faststream.asgi.types import Scope (empty)

7from faststream.specification.asyncapi.site import ( (empty)

8 ASYNCAPI_CSS_DEFAULT_URL, 

9 ASYNCAPI_JS_DEFAULT_URL, 

10 ASYNCAPI_TRY_IT_PLUGIN_URL, 

11 get_asyncapi_html, 

12) 

13 

14if TYPE_CHECKING: (empty)

15 from faststream.specification.base import SpecificationFactory 

16 from faststream.specification.schema import Tag, TagDict 

17 

18 

19def make_asyncapi_asgi( (empty)

20 schema: "SpecificationFactory", 

21 description: str | None = None, 

22 tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None, 

23 unique_id: str | None = None, 

24 include_in_schema: bool = True, 

25 *, 

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) -> "GetHandler": 

40 """Create AsyncAPI documentation ASGI handler.""" 

41 cached_docs: str | None = None (empty)

42 

43 @get( (empty)

44 include_in_schema=include_in_schema, 

45 description=description, 

46 tags=tags, 

47 unique_id=unique_id, 

48 ) 

49 async def docs(scope: Scope) -> AsgiResponse: (empty)

50 nonlocal cached_docs 

51 if not cached_docs: 51 ↛ 69line 51 didn't jump to line 69 because the condition on line 51 was always true(empty)

52 cached_docs = get_asyncapi_html( (empty)

53 schema.to_specification(), 

54 sidebar=sidebar, 

55 info=info, 

56 servers=servers, 

57 operations=operations, 

58 messages=messages, 

59 schemas=schemas, 

60 errors=errors, 

61 expand_message_examples=expand_message_examples, 

62 asyncapi_js_url=asyncapi_js_url, 

63 asyncapi_css_url=asyncapi_css_url, 

64 try_it_out_plugin_url=try_it_out_plugin_url, 

65 try_it_out=try_it_out, 

66 try_it_out_url=try_it_out_url, 

67 ) 

68 

69 return AsgiResponse( (empty)

70 cached_docs.encode("utf-8"), 

71 200, 

72 {"Content-Type": "text/html; charset=utf-8"}, 

73 ) 

74 

75 return docs (empty)