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

1from collections.abc import Sequence (empty)

2from contextlib import suppress (empty)

3from functools import lru_cache (empty)

4from typing import TYPE_CHECKING, Any, TypedDict, Union (empty)

5 

6from faststream.asgi.annotations import Request (empty)

7from faststream.asgi.handlers import PostHandler, post (empty)

8from faststream.asgi.response import AsgiResponse, JSONResponse (empty)

9from faststream.exceptions import SubscriberNotFound (empty)

10 

11if TYPE_CHECKING: (empty)

12 from faststream._internal.broker import BrokerUsecase 

13 from faststream._internal.testing.broker import TestBroker 

14 from faststream.specification.schema import Tag, TagDict 

15 

16 

17class TryItOutOptions(TypedDict, total=False): (empty)

18 sendToRealBroker: bool (empty)

19 timestamp: str (empty)

20 

21 

22class TryItOutMessage(TypedDict, total=False): (empty)

23 """Wrapper sent by asyncapi-try-it-plugin. (empty)

24 

25 The plugin always wraps the user's payload inside a nested ``message`` 

26 field together with operation metadata:: 

27 

28 { 

29 "operation_id": "...", 

30 "operation_type": "...", 

31 "message": <actual_user_payload> 

32 } 

33 """ 

34 

35 operation_id: str (empty)

36 operation_type: str (empty)

37 message: Any (empty)

38 

39 

40class TryItOutForm(TypedDict): (empty)

41 channelName: str (empty)

42 message: TryItOutMessage (empty)

43 options: TryItOutOptions (empty)

44 

45 

46class TryItOutProcessor: (empty)

47 """Process try-it-out requests: parse, validate, publish to real or test broker.""" 

48 

49 def __init__(self, broker: "BrokerUsecase[Any, Any]") -> None: (empty)

50 self._broker = broker (empty)

51 

52 registry = _get_broker_registry() (empty)

53 for br_cls, test_broker_cls in registry.items(): 53 ↛ 59line 53 didn't jump to line 59 because the loop on line 53 didn't complete(empty)

54 if isinstance(self._broker, br_cls): (empty)

55 self._test_broker_cls = test_broker_cls (empty)

56 break (empty)

57 

58 else: 

59 msg = f"TestBroker not available for {broker}. Please, inspect your dependencies." 

60 raise ValueError(msg) 

61 

62 async def process(self, body: TryItOutForm) -> AsgiResponse: (empty)

63 """Process parsed body: validate, dry-run or publish. Returns response.""" 

64 destination, *_ = body.get("channelName", "").split(":") (empty)

65 

66 if not destination: (empty)

67 return JSONResponse({"details": "Missing channelName"}, 400) (empty)

68 

69 message_wrapper = body.get("message", {}) (empty)

70 payload: Any = message_wrapper.get("message") (empty)

71 options = body.get("options", {}) (empty)

72 use_real_broker = options.get("sendToRealBroker", False) (empty)

73 

74 try: (empty)

75 if use_real_broker: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true(empty)

76 await self._broker.publish(payload, destination) 

77 return JSONResponse("ok", 200) 

78 

79 async with self._test_broker_cls(self._broker) as br: (empty)

80 data = await br.request(payload, destination, timeout=30) (empty)

81 decoded = None (empty)

82 with suppress(Exception): (empty)

83 decoded = await data.decode() (empty)

84 return JSONResponse( (empty)

85 decoded if decoded is not None and decoded != b"" else "ok", 200 

86 ) 

87 

88 except SubscriberNotFound: (empty)

89 return JSONResponse({"details": f"{destination} destination not found."}, 404) (empty)

90 

91 except Exception as e: (empty)

92 return JSONResponse({"details": repr(e)}, 500) (empty)

93 

94 

95def make_try_it_out_handler( (empty)

96 broker: "BrokerUsecase[Any, Any]", 

97 description: str | None = None, 

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

99 unique_id: str | None = None, 

100 include_in_schema: bool = False, 

101) -> "PostHandler": 

102 """Create POST handler for asyncapi-try-it-plugin to publish messages to broker.""" 

103 processor = TryItOutProcessor(broker) (empty)

104 

105 @post( (empty)

106 description=description, 

107 tags=tags, 

108 unique_id=unique_id, 

109 include_in_schema=include_in_schema, 

110 ) 

111 async def try_it_out(request: Request) -> AsgiResponse: (empty)

112 try: (empty)

113 body: TryItOutForm = await request.json() (empty)

114 

115 except Exception as e: 

116 return JSONResponse({"details": f"Invalid JSON: {e}"}, 400) 

117 

118 return await processor.process(body) (empty)

119 

120 return try_it_out (empty)

121 

122 

123@lru_cache(maxsize=1) (empty)

124def _get_broker_registry() -> dict[ (empty)

125 type["BrokerUsecase[Any, Any]"], 

126 type["TestBroker[Any]"], 

127]: 

128 registry: dict[type[BrokerUsecase[Any, Any]], type[TestBroker[Any]]] = {} (empty)

129 

130 with suppress(ImportError): (empty)

131 from faststream.confluent import ( (empty)

132 KafkaBroker as ConfluentKafkaBroker, 

133 TestKafkaBroker as TestConfluentKafkaBroker, 

134 ) 

135 

136 registry[ConfluentKafkaBroker] = TestConfluentKafkaBroker (empty)

137 

138 with suppress(ImportError): (empty)

139 from faststream.kafka import ( (empty)

140 KafkaBroker as AioKafkaBroker, 

141 TestKafkaBroker as TestAioKafkaBroker, 

142 ) 

143 

144 registry[AioKafkaBroker] = TestAioKafkaBroker (empty)

145 

146 with suppress(ImportError): (empty)

147 from faststream.nats import NatsBroker, TestNatsBroker (empty)

148 

149 registry[NatsBroker] = TestNatsBroker (empty)

150 

151 with suppress(ImportError): (empty)

152 from faststream.rabbit import RabbitBroker, TestRabbitBroker (empty)

153 

154 registry[RabbitBroker] = TestRabbitBroker (empty)

155 

156 with suppress(ImportError): (empty)

157 from faststream.redis import RedisBroker, TestRedisBroker (empty)

158 

159 registry[RedisBroker] = TestRedisBroker (empty)

160 

161 with suppress(ImportError): (empty)

162 from faststream.mqtt import MQTTBroker, TestMQTTBroker (empty)

163 

164 registry[MQTTBroker] = TestMQTTBroker (empty)

165 

166 return registry (empty)