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

1# from __future__ import annotations 

2 

3from datetime import datetime, timedelta 

4from typing import ( 

5 TYPE_CHECKING, 

6 Any, 

7 Awaitable, 

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 

22Number = Union[int, float] 

23 

24 

25class AsyncClientProtocol(Protocol): 

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

27 

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

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

30 """ 

31 

32 _is_async_client: Literal[True] 

33 

34 

35class SyncClientProtocol(Protocol): 

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

37 

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

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

40 """ 

41 

42 _is_async_client: Literal[False] 

43 

44 

45EncodedT = Union[bytes, bytearray, memoryview] 

46DecodedT = Union[str, int, float] 

47EncodableT = Union[EncodedT, DecodedT] 

48AbsExpiryT = Union[int, datetime] 

49ExpiryT = Union[int, timedelta] 

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

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

52_StringLikeT = Union[bytes, str, memoryview] 

53KeyT = _StringLikeT # Main redis key space 

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

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

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

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

58ChannelT = _StringLikeT 

59GroupT = _StringLikeT # Consumer group 

60ConsumerT = _StringLikeT # Consumer name 

61StreamIdT = Union[int, _StringLikeT] 

62ScriptTextT = _StringLikeT 

63TimeoutSecT = Union[int, float, _StringLikeT] 

64ACLGetUserData = ( 

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

66) 

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

68ACLLogData = list[ACLLogEntry] 

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

70CommandGetKeysAndFlagsResponse = list[CommandGetKeysAndFlagsEntry] 

71BlockingListPopResponse = tuple[bytes | str, bytes | str] | 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 

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

78StreamRangeResponse = list[StreamEntry] 

79XClaimResponse = StreamRangeResponse | list[bytes | str] 

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

81XPendingRangeResponse = list[XPendingRangeEntry] 

82XReadResponse = list[list[Any]] | dict[bytes | str, list[StreamRangeResponse]] 

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

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

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

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

87TimeSeriesRangeResponse = list[TimeSeriesSample] 

88BloomScanDumpResponse = tuple[int, bytes | None] 

89ModuleListResponse = list[int | float | str | None] 

90BlockingZSetPopResponse = ( 

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

92) 

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

94ZRandMemberResponse = ( 

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

96) 

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

98ZSetRangeResponse = list[bytes | str] | ZSetScoredMembers 

99ZScanResponse = tuple[int, list[tuple[bytes | str, float]]] 

100LCSMatch = list[int | tuple[int, int]] 

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

102StralgoResponse = str | int | LCSResult 

103 

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

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

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

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

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

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

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

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

112 

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

114 

115 

116class CommandsProtocol(Protocol): 

117 _event_dispatcher: "EventDispatcherInterface" 

118 

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

120 

121 

122class ClusterCommandsProtocol(CommandsProtocol): 

123 encoder: "Encoder"