Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/redis/typing.py: 98%

1from dataclasses import dataclass 

2from datetime import datetime, timedelta 

3from typing import ( 

4 TYPE_CHECKING, 

5 Any, 

6 Awaitable, 

7 Callable, 

8 Iterable, 

9 Literal, 

10 Mapping, 

11 Protocol, 

12 Type, 

13 TypeVar, 

14 Union, 

15) 

16 

17if TYPE_CHECKING: 

18 from redis._parsers import Encoder 

19 from redis.event import EventDispatcherInterface 

20 

21 

22class AsyncClientProtocol(Protocol): 

23 """Protocol for asynchronous Redis clients (redis.asyncio.client.Redis). 

24 

25 This protocol uses a Literal marker to identify async clients. 

26 Used in @overload to provide correct return types for async clients. 

27 """ 

28 

29 _is_async_client: Literal[True] 

30 

31 

32class SyncClientProtocol(Protocol): 

33 """Protocol for synchronous Redis clients (redis.client.Redis). 

34 

35 This protocol uses a Literal marker to identify sync clients. 

36 Used in @overload to provide correct return types for sync clients. 

37 """ 

38 

39 _is_async_client: Literal[False] 

40 

41 

42Number = Union[int, float] 

43 

44EncodedT = Union[bytes, bytearray, memoryview] 

45DecodedT = Union[str, int, float] 

46EncodableT = Union[EncodedT, DecodedT] 

47AbsExpiryT = Union[int, datetime] 

48ExpiryT = Union[int, timedelta] 

49ZScoreBoundT = Union[float, str] # str allows for the [ or ( prefix 

50BitfieldOffsetT = Union[int, str] # str allows for #x syntax 

51_StringLikeT = Union[bytes, str, memoryview] 

52KeyT = _StringLikeT # Main redis key space 

53PatternT = _StringLikeT # Patterns matched against keys, fields etc 

54FieldT = EncodableT # Fields within hash tables, streams and geo commands 

55KeysT = Union[KeyT, Iterable[KeyT]] 

56ResponseT = Union[Awaitable[Any], Any] 

57GroupT = _StringLikeT # Consumer group 

58ConsumerT = _StringLikeT # Consumer name 

59StreamIdT = Union[int, _StringLikeT] 

60ScriptTextT = _StringLikeT 

61TimeoutSecT = Union[int, float, _StringLikeT] 

62ACLGetUserData = ( 

63 dict[str, bool | list[str] | list[list[str]] | list[dict[str, str]]] | None 

64) 

65ACLLogEntry = dict[str, str | float | dict[str, str | int]] 

66ACLLogData = list[ACLLogEntry] 

67CommandGetKeysAndFlagsEntry = list[bytes | str | list[bytes | str]] 

68CommandGetKeysAndFlagsResponse = list[CommandGetKeysAndFlagsEntry] 

69ClientTrackingInfoResponse = list[bytes | str] | dict[str, Any] 

70BlockingListPopResponse = tuple[bytes | str, bytes | str] | list[bytes | str] | None 

71HRandFieldResponse = bytes | str | list[bytes | str] | list[list[bytes | str]] | None 

72HScanPayload = dict[bytes | str, bytes | str] | list[bytes | str] 

73HScanResponse = tuple[int, HScanPayload] 

74ListMultiPopResponse = list[bytes | str | list[bytes | str]] | None 

75ScanResponse = tuple[int, list[bytes | str]] 

76SortResponse = list[bytes | str] | list[tuple[bytes | str, ...]] | int 

77GeoCoordinate = tuple[float, float] | list[float] 

78GeoSearchItem = bytes | str | list[bytes | str | float | int | GeoCoordinate] 

79GeoSearchResponse = list[GeoSearchItem] 

80GeoRadiusResponse = GeoSearchResponse | int 

81StreamEntry = tuple[bytes | str | None, dict[bytes | str, bytes | str] | None] 

82StreamRangeResponse = list[StreamEntry] 

83XClaimResponse = StreamRangeResponse | list[bytes | str] 

84XPendingRangeEntry = dict[str, bytes | str | int] 

85XPendingRangeResponse = list[XPendingRangeEntry] 

86XReadGroupClaimEntry = tuple[ 

87 bytes | str, 

88 dict[bytes | str, bytes | str], 

89 bytes | str | int, 

90 bytes | str | int, 

91] 

92XReadGroupStreamResponse = StreamRangeResponse | list[XReadGroupClaimEntry] 

93XReadResponse = ( 

94 list[list[Any]] 

95 | dict[bytes | str, list[StreamRangeResponse]] 

96 | dict[bytes | str, StreamRangeResponse] 

97) 

98XReadGroupResponse = ( 

99 list[list[Any]] 

100 | dict[bytes | str, list[XReadGroupStreamResponse]] 

101 | dict[bytes | str, XReadGroupStreamResponse] 

102) 

103ClusterNodeDetail = dict[str, str | bool | list[list[str]] | list[dict[str, str]]] 

104ClusterLink = dict[str, Any] | list[Any] 

105ClusterLinksResponse = list[ClusterLink] 

106ClusterShard = dict[str, Any] 

107ClusterShardsResponse = list[ClusterShard] 

108SentinelMasterAddress = tuple[bytes | str, int] | None 

109SentinelMastersResponse = dict[str, dict[str, Any]] | list[dict[str, Any]] 

110TimeSeriesSample = tuple[int, float] | list[int | float] 

111TimeSeriesRangeResponse = list[TimeSeriesSample] 

112TimeSeriesMRangeSeries = list[Any] 

113TimeSeriesMRangeResponse = list[Any] | dict[bytes | str, TimeSeriesMRangeSeries] 

114BloomScanDumpResponse = tuple[int, bytes | None] 

115ModuleListResponse = list[bytes | int | float | str | None] 

116BlockingZSetPopResponse = ( 

117 tuple[bytes | str, bytes | str, float] | list[bytes | str | float] | None 

118) 

119ZMPopResponse = list[bytes | str | list[list[Any]]] | None 

120ZRandMemberResponse = ( 

121 bytes | str | None | list[bytes | str] | list[bytes | str | float] | list[list[Any]] 

122) 

123ZSetScoredMembers = list[tuple[bytes | str, Any]] | list[list[Any]] 

124ZSetRangeResponse = list[bytes | str] | ZSetScoredMembers 

125ZScanPair = tuple[bytes | str, float] | list[bytes | str | float] 

126ZScanResponse = tuple[int, list[ZScanPair]] 

127LCSRange = tuple[int, int] | list[int] 

128LCSMatch = list[int | LCSRange] 

129LCSResult = dict[str, int | list[LCSMatch]] 

130LCSIndexResponse = list[Any] | dict[bytes | str, Any] 

131LCSCommandResponse = bytes | str | int | LCSIndexResponse 

132StralgoResponse = str | int | LCSResult 

133 

134# Mapping is not covariant in the key type, which prevents 

135# Mapping[_StringLikeT, X] from accepting arguments of type Dict[str, X]. Using 

136# a TypeVar instead of a Union allows mappings with any of the permitted types 

137# to be passed. Care is needed if there is more than one such mapping in a 

138# type signature because they will all be required to be the same key type. 

139AnyKeyT = TypeVar("AnyKeyT", bytes, str, memoryview) 

140AnyFieldT = TypeVar("AnyFieldT", bytes, str, memoryview) 

141 

142ExceptionMappingT = Mapping[str, Union[Type[Exception], Mapping[str, Type[Exception]]]] 

143 

144ChannelT = _StringLikeT 

145AnyChannelT = TypeVar("AnyChannelT", bytes, str, memoryview) 

146PubSubHandler = Callable[[dict[str, Any]], Any] 

147 

148 

149@dataclass(frozen=True) 

150class Subscription: 

151 """PubSub channel or pattern subscription with an optional handler.""" 

152 

153 name: ChannelT 

154 handler: PubSubHandler | None = None 

155 

156 

157class CommandsProtocol(Protocol): 

158 _event_dispatcher: "EventDispatcherInterface" 

159 

160 def execute_command(self, *args, **options) -> ResponseT: ... 

161 

162 

163class ClusterCommandsProtocol(CommandsProtocol): 

164 encoder: "Encoder"