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 = list[bytes | str] | None 

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

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

74HScanResponse = tuple[int, HScanPayload] 

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

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

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

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

79StreamRangeResponse = list[StreamEntry] 

80XClaimResponse = StreamRangeResponse | list[bytes | str] 

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

82XPendingRangeResponse = list[XPendingRangeEntry] 

83XReadResponse = dict[bytes | str, list[StreamRangeResponse]] 

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

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

86SentinelMastersResponse = dict[str, dict[str, Any]] 

87TimeSeriesSample = list[int | float] 

88TimeSeriesRangeResponse = list[TimeSeriesSample] 

89BloomScanDumpResponse = tuple[int, bytes | None] 

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

91BlockingZSetPopResponse = list[bytes | str | float] | None 

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

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

94ZSetScoredMembers = list[list[Any]] 

95ZSetRangeResponse = list[bytes | str] | ZSetScoredMembers 

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

97LCSMatch = list[int | list[int]] 

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

99StralgoResponse = str | int | LCSResult 

100 

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

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

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

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

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

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

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

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

109 

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

111 

112 

113class CommandsProtocol(Protocol): 

114 _event_dispatcher: "EventDispatcherInterface" 

115 

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

117 

118 

119class ClusterCommandsProtocol(CommandsProtocol): 

120 encoder: "Encoder"