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

1# from __future__ import annotations 

2 

3from datetime import datetime, timedelta 

4from typing import ( 

5 TYPE_CHECKING, 

6 Any, 

7 Awaitable, 

8 Iterable, 

9 Mapping, 

10 Protocol, 

11 Type, 

12 TypeVar, 

13 Union, 

14) 

15 

16if TYPE_CHECKING: 

17 from redis._parsers import Encoder 

18 from redis.event import EventDispatcherInterface 

19 

20 

21Number = Union[int, float] 

22EncodedT = Union[bytes, bytearray, memoryview] 

23DecodedT = Union[str, int, float] 

24EncodableT = Union[EncodedT, DecodedT] 

25AbsExpiryT = Union[int, datetime] 

26ExpiryT = Union[int, timedelta] 

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

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

29_StringLikeT = Union[bytes, str, memoryview] 

30KeyT = _StringLikeT # Main redis key space 

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

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

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

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

35ChannelT = _StringLikeT 

36GroupT = _StringLikeT # Consumer group 

37ConsumerT = _StringLikeT # Consumer name 

38StreamIdT = Union[int, _StringLikeT] 

39ScriptTextT = _StringLikeT 

40TimeoutSecT = Union[int, float, _StringLikeT] 

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

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

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

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

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

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

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

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

49 

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

51 

52 

53class CommandsProtocol(Protocol): 

54 _event_dispatcher: "EventDispatcherInterface" 

55 

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

57 

58 

59class ClusterCommandsProtocol(CommandsProtocol): 

60 encoder: "Encoder"