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

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 

19 

20Number = Union[int, float] 

21EncodedT = Union[bytes, bytearray, memoryview] 

22DecodedT = Union[str, int, float] 

23EncodableT = Union[EncodedT, DecodedT] 

24AbsExpiryT = Union[int, datetime] 

25ExpiryT = Union[int, timedelta] 

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

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

28_StringLikeT = Union[bytes, str, memoryview] 

29KeyT = _StringLikeT # Main redis key space 

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

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

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

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

34ChannelT = _StringLikeT 

35GroupT = _StringLikeT # Consumer group 

36ConsumerT = _StringLikeT # Consumer name 

37StreamIdT = Union[int, _StringLikeT] 

38ScriptTextT = _StringLikeT 

39TimeoutSecT = Union[int, float, _StringLikeT] 

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

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

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

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

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

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

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

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

48 

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

50 

51 

52class CommandsProtocol(Protocol): 

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

54 

55 

56class ClusterCommandsProtocol(CommandsProtocol): 

57 encoder: "Encoder"