Coverage for faststream / asgi / response.py: 97%

1from collections.abc import Mapping (empty)

2from typing import TYPE_CHECKING, Any (empty)

3 

4from faststream._internal._compat import json_dumps (empty)

5 

6if TYPE_CHECKING: (empty)

7 from .types import Receive, Scope, Send 

8 

9 

10class AsgiResponse: (empty)

11 def __init__( (empty)

12 self, 

13 body: bytes = b"", 

14 status_code: int = 200, 

15 headers: Mapping[str, str] | None = None, 

16 ) -> None: 

17 self.status_code = status_code (empty)

18 self.body = body (empty)

19 self.raw_headers = _get_response_headers(body, headers, status_code) (empty)

20 

21 def __repr__(self) -> str: (empty)

22 inner = [f"status_code={self.status_code}"] 

23 if (ln := len(self.body)) > 100: 

24 inner.append( 

25 f"body={self.body[:100]!r}".rstrip(" \\n'") + f" ...' ({ln} bytes)" 

26 ) 

27 else: 

28 inner.append(f"body={self.body!r}") 

29 inner.append(f"headers={self.raw_headers}") 

30 return f"{self.__class__.__name__}({', '.join(inner)})" 

31 

32 async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None: (empty)

33 prefix = "websocket." if (scope["type"] == "websocket") else "" (empty)

34 await send( (empty)

35 { 

36 "type": f"{prefix}http.response.start", 

37 "status": self.status_code, 

38 "headers": self.raw_headers, 

39 }, 

40 ) 

41 await send( (empty)

42 { 

43 "type": f"{prefix}http.response.body", 

44 "body": self.body, 

45 }, 

46 ) 

47 

48 

49def JSONResponse( # noqa: N802 (empty)

50 data: Any, 

51 status_code: int = 200, 

52 headers: Mapping[str, str] | None = None, 

53) -> AsgiResponse: 

54 if not isinstance(data, bytes): 54 ↛ 57line 54 didn't jump to line 57 because the condition on line 54 was always true(empty)

55 data = json_dumps(data) (empty)

56 

57 return AsgiResponse( (empty)

58 data, 

59 status_code, 

60 {"Content-Type": "application/json", **(headers or {})}, 

61 ) 

62 

63 

64def _get_response_headers( (empty)

65 body: bytes, 

66 headers: Mapping[str, str] | None, 

67 status_code: int, 

68) -> list[tuple[bytes, bytes]]: 

69 if headers is None: (empty)

70 raw_headers: list[tuple[bytes, bytes]] = [] (empty)

71 populate_content_length = True (empty)

72 

73 else: 

74 raw_headers = [ (empty)

75 (k.lower().encode("latin-1"), v.encode("latin-1")) for k, v in headers.items() 

76 ] 

77 keys = [h[0] for h in raw_headers] (empty)

78 populate_content_length = b"content-length" not in keys (empty)

79 

80 if ( (empty)

81 body 

82 and populate_content_length 

83 and not (status_code < 200 or status_code in {204, 304}) 

84 ): 

85 content_length = str(len(body)) (empty)

86 raw_headers.append((b"content-length", content_length.encode("latin-1"))) (empty)

87 

88 return raw_headers (empty)