Coverage for faststream / redis / publisher / specification.py: 93%

1from faststream._internal.endpoint.publisher import PublisherSpecification (empty)

2from faststream.redis.configs import RedisBrokerConfig (empty)

3from faststream.redis.schemas import ListSub, PubSub, StreamSub (empty)

4from faststream.specification.asyncapi.utils import resolve_payloads (empty)

5from faststream.specification.schema import Message, Operation, PublisherSpec (empty)

6from faststream.specification.schema.bindings import ChannelBinding, redis (empty)

7 

8from .config import RedisPublisherSpecificationConfig (empty)

9 

10 

11class RedisPublisherSpecification( (empty)

12 PublisherSpecification[RedisBrokerConfig, RedisPublisherSpecificationConfig], 

13): 

14 def get_schema(self) -> dict[str, PublisherSpec]: (empty)

15 payloads = self.get_payloads() (empty)

16 

17 return { (empty)

18 self.name: PublisherSpec( 

19 description=self.config.description_, 

20 operation=Operation( 

21 message=Message( 

22 title=f"{self.name}:Message", 

23 payload=resolve_payloads(payloads, "Publisher"), 

24 ), 

25 bindings=None, 

26 ), 

27 bindings=ChannelBinding( 

28 redis=self.channel_binding, 

29 ), 

30 ), 

31 } 

32 

33 @property (empty)

34 def channel_binding(self) -> redis.ChannelBinding: (empty)

35 raise NotImplementedError 

36 

37 

38class ChannelPublisherSpecification(RedisPublisherSpecification): (empty)

39 def __init__( (empty)

40 self, 

41 _outer_config: RedisBrokerConfig, 

42 specification_config: RedisPublisherSpecificationConfig, 

43 channel: PubSub, 

44 ) -> None: 

45 super().__init__(_outer_config, specification_config) (empty)

46 self.channel = channel (empty)

47 

48 @property (empty)

49 def name(self) -> str: (empty)

50 if self.config.title_: (empty)

51 return self.config.title_ (empty)

52 

53 return f"{self.channel_name}:Publisher" (empty)

54 

55 @property (empty)

56 def channel_name(self) -> str: (empty)

57 return f"{self._outer_config.prefix}{self.channel.name}" (empty)

58 

59 @property (empty)

60 def channel_binding(self) -> redis.ChannelBinding: (empty)

61 return redis.ChannelBinding( (empty)

62 channel=self.channel_name, 

63 method="publish", 

64 ) 

65 

66 

67class ListPublisherSpecification(RedisPublisherSpecification): (empty)

68 def __init__( (empty)

69 self, 

70 _outer_config: RedisBrokerConfig, 

71 specification_config: RedisPublisherSpecificationConfig, 

72 list_sub: ListSub, 

73 ) -> None: 

74 super().__init__(_outer_config, specification_config) (empty)

75 self.list_sub = list_sub (empty)

76 

77 @property (empty)

78 def name(self) -> str: (empty)

79 if self.config.title_: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true(empty)

80 return self.config.title_ 

81 

82 return f"{self.list_name}:Publisher" (empty)

83 

84 @property (empty)

85 def list_name(self) -> str: (empty)

86 return f"{self._outer_config.prefix}{self.list_sub.name}" (empty)

87 

88 @property (empty)

89 def channel_binding(self) -> redis.ChannelBinding: (empty)

90 return redis.ChannelBinding( (empty)

91 channel=self.list_name, 

92 method="rpush", 

93 ) 

94 

95 

96class StreamPublisherSpecification(RedisPublisherSpecification): (empty)

97 def __init__( (empty)

98 self, 

99 _outer_config: RedisBrokerConfig, 

100 specification_config: RedisPublisherSpecificationConfig, 

101 stream_sub: StreamSub, 

102 ) -> None: 

103 super().__init__(_outer_config, specification_config) (empty)

104 self.stream_sub = stream_sub (empty)

105 

106 @property (empty)

107 def name(self) -> str: (empty)

108 if self.config.title_: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true(empty)

109 return self.config.title_ 

110 

111 return f"{self.stream_name}:Publisher" (empty)

112 

113 @property (empty)

114 def stream_name(self) -> str: (empty)

115 return f"{self._outer_config.prefix}{self.stream_sub.name}" (empty)

116 

117 @property (empty)

118 def channel_binding(self) -> "redis.ChannelBinding": (empty)

119 return redis.ChannelBinding( (empty)

120 channel=self.stream_name, 

121 method="xadd", 

122 )