Coverage for faststream / _internal / broker / pub_base.py: 93%

1from abc import abstractmethod (empty)

2from collections.abc import Sequence (empty)

3from functools import partial (empty)

4from typing import TYPE_CHECKING, Any, Generic (empty)

5 

6from faststream._internal.endpoint.utils import process_msg (empty)

7from faststream._internal.types import MsgType (empty)

8from faststream.exceptions import FeatureNotSupportedException (empty)

9from faststream.message.source_type import SourceType (empty)

10 

11if TYPE_CHECKING: (empty)

12 from faststream._internal.basic_types import SendableMessage 

13 from faststream._internal.context import ContextRepo 

14 from faststream._internal.producer import ProducerProto 

15 from faststream._internal.types import BrokerMiddleware 

16 from faststream.response import PublishCommand 

17 

18 

19class BrokerPublishMixin(Generic[MsgType]): (empty)

20 @property (empty)

21 @abstractmethod (empty)

22 def middlewares(self) -> Sequence["BrokerMiddleware[MsgType]"]: (empty)

23 raise NotImplementedError 

24 

25 @property (empty)

26 @abstractmethod (empty)

27 def context(self) -> "ContextRepo": (empty)

28 raise NotImplementedError 

29 

30 @abstractmethod (empty)

31 async def publish( (empty)

32 self, 

33 message: "SendableMessage", (empty)

34 queue: str, (empty)

35 /, 

36 ) -> Any: (empty)

37 raise NotImplementedError 

38 

39 async def _basic_publish( (empty)

40 self, 

41 cmd: "PublishCommand", 

42 *, 

43 producer: "ProducerProto[Any]", 

44 ) -> Any: 

45 publish = producer.publish (empty)

46 context = self.context # caches property (empty)

47 

48 for m in self.middlewares[::-1]: (empty)

49 publish = partial(m(None, context=context).publish_scope, publish) (empty)

50 

51 return await publish(cmd) (empty)

52 

53 async def publish_batch( (empty)

54 self, 

55 *messages: "SendableMessage", 

56 queue: str, 

57 ) -> Any: 

58 msg = f"{self.__class__.__name__} doesn't support publishing in batches." 

59 raise FeatureNotSupportedException(msg) 

60 

61 async def _basic_publish_batch( (empty)

62 self, 

63 cmd: "PublishCommand", 

64 *, 

65 producer: "ProducerProto[Any]", 

66 ) -> Any: 

67 publish = producer.publish_batch (empty)

68 context = self.context # caches property (empty)

69 

70 for m in self.middlewares[::-1]: (empty)

71 publish = partial(m(None, context=context).publish_scope, publish) (empty)

72 

73 return await publish(cmd) (empty)

74 

75 @abstractmethod (empty)

76 async def request( (empty)

77 self, 

78 message: "SendableMessage", (empty)

79 queue: str, (empty)

80 /, 

81 timeout: float = 0.5, (empty)

82 ) -> Any: (empty)

83 raise NotImplementedError 

84 

85 async def _basic_request( (empty)

86 self, 

87 cmd: "PublishCommand", 

88 *, 

89 producer: "ProducerProto[Any]", 

90 ) -> Any: 

91 request = producer.request (empty)

92 context = self.context # caches property (empty)

93 

94 for m in self.middlewares[::-1]: (empty)

95 request = partial(m(None, context=context).publish_scope, request) (empty)

96 

97 published_msg = await request(cmd) (empty)

98 

99 response_msg: Any = await process_msg( (empty)

100 msg=published_msg, 

101 middlewares=(m(published_msg, context=context) for m in self.middlewares), 

102 parser=producer._parser, 

103 decoder=producer._decoder, 

104 source_type=SourceType.RESPONSE, 

105 ) 

106 return response_msg (empty)