Coverage for faststream / redis / message.py: 97%

1from typing import ( (empty)

2 TYPE_CHECKING, 

3 Literal, 

4 Optional, 

5 TypeAlias, 

6 TypeVar, 

7 Union, 

8) 

9 

10from typing_extensions import NotRequired, TypedDict, override (empty)

11 

12from faststream.message import StreamMessage as BrokerStreamMessage (empty)

13 

14if TYPE_CHECKING: (empty)

15 from redis.asyncio import Redis 

16 

17 from faststream._internal.basic_types import DecodedMessage 

18 

19 

20BaseMessage: TypeAlias = Union[ (empty)

21 "PubSubMessage", 

22 "DefaultListMessage", 

23 "BatchListMessage", 

24 "DefaultStreamMessage", 

25 "BatchStreamMessage", 

26] 

27 

28 

29class UnifyRedisDict(TypedDict): (empty)

30 type: Literal[ (empty)

31 "pmessage", 

32 "message", 

33 "list", 

34 "blist", 

35 "stream", 

36 "bstream", 

37 ] 

38 channel: str (empty)

39 data: bytes | list[bytes] | dict[bytes, bytes] | list[dict[bytes, bytes]] (empty)

40 pattern: NotRequired[bytes | None] (empty)

41 

42 

43class RedisMessage(BrokerStreamMessage[UnifyRedisDict]): (empty)

44 pass (empty)

45 

46 

47class PubSubMessage(TypedDict): (empty)

48 """A class to represent a PubSub message.""" 

49 

50 type: Literal["pmessage", "message"] (empty)

51 channel: str (empty)

52 data: bytes (empty)

53 pattern: bytes | None (empty)

54 

55 

56class RedisChannelMessage(BrokerStreamMessage[PubSubMessage]): (empty)

57 pass (empty)

58 

59 

60class _ListMessage(TypedDict): (empty)

61 """A class to represent an Abstract List message.""" 

62 

63 channel: str (empty)

64 

65 

66class DefaultListMessage(_ListMessage): (empty)

67 """A class to represent a single List message.""" 

68 

69 type: Literal["list"] (empty)

70 data: bytes (empty)

71 

72 

73class BatchListMessage(_ListMessage): (empty)

74 """A class to represent a List messages batch.""" 

75 

76 type: Literal["blist"] (empty)

77 data: list[bytes] (empty)

78 

79 

80class RedisListMessage(BrokerStreamMessage[DefaultListMessage]): (empty)

81 """StreamMessage for single List message.""" 

82 

83 

84class RedisBatchListMessage(BrokerStreamMessage[BatchListMessage]): (empty)

85 """StreamMessage for single List message.""" 

86 

87 decoded_body: list["DecodedMessage"] (empty)

88 

89 

90DATA_KEY = "__data__" (empty)

91bDATA_KEY = DATA_KEY.encode() # noqa: N816 (empty)

92 

93 

94class _StreamMessage(TypedDict): (empty)

95 channel: str (empty)

96 message_ids: list[bytes] (empty)

97 

98 

99class DefaultStreamMessage(_StreamMessage): (empty)

100 type: Literal["stream"] (empty)

101 data: dict[bytes, bytes] (empty)

102 

103 

104class BatchStreamMessage(_StreamMessage): (empty)

105 type: Literal["bstream"] (empty)

106 data: list[dict[bytes, bytes]] (empty)

107 

108 

109_StreamMsgType = TypeVar("_StreamMsgType", bound=_StreamMessage) (empty)

110 

111 

112class _RedisStreamMessageMixin(BrokerStreamMessage[_StreamMsgType]): (empty)

113 @override (empty)

114 async def ack( (empty)

115 self, 

116 redis: Optional["Redis[bytes]"] = None, 

117 group: str | None = None, 

118 ) -> None: 

119 if not self.committed and group is not None and redis is not None: (empty)

120 ids = self.raw_message["message_ids"] (empty)

121 channel = self.raw_message["channel"] (empty)

122 await redis.xack(channel, group, *ids) # type: ignore[no-untyped-call] (empty)

123 await super().ack() (empty)

124 

125 @override (empty)

126 async def nack( (empty)

127 self, 

128 redis: Optional["Redis[bytes]"] = None, 

129 group: str | None = None, 

130 ) -> None: 

131 await super().nack() (empty)

132 

133 @override (empty)

134 async def reject( (empty)

135 self, 

136 redis: Optional["Redis[bytes]"] = None, 

137 group: str | None = None, 

138 ) -> None: 

139 await super().reject() 

140 

141 async def delete(self, redis: Optional["Redis[bytes]"]) -> None: (empty)

142 if redis is not None: 142 ↛ exitline 142 didn't return from function 'delete' because the condition on line 142 was always true(empty)

143 ids = self.raw_message["message_ids"] (empty)

144 channel = self.raw_message["channel"] (empty)

145 await redis.xdel(channel, *ids) (empty)

146 

147 

148class RedisStreamMessage(_RedisStreamMessageMixin[DefaultStreamMessage]): (empty)

149 pass (empty)

150 

151 

152class RedisBatchStreamMessage(_RedisStreamMessageMixin[BatchStreamMessage]): (empty)

153 decoded_body: list["DecodedMessage"] (empty)