Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/limits/typing.py: 2%

1from typing import ( 

2 TYPE_CHECKING, 

3 Callable, 

4 Dict, 

5 List, 

6 NamedTuple, 

7 Optional, 

8 Tuple, 

9 Type, 

10 TypeVar, 

11 Union, 

12) 

13 

14from typing_extensions import ClassVar, Counter, ParamSpec, Protocol 

15 

16Serializable = Union[int, str, float] 

17 

18R = TypeVar("R") 

19R_co = TypeVar("R_co", covariant=True) 

20P = ParamSpec("P") 

21 

22 

23if TYPE_CHECKING: 

24 import coredis 

25 import coredis.commands.script 

26 import redis 

27 

28 

29class ItemP(Protocol): 

30 value: bytes 

31 flags: Optional[int] 

32 cas: Optional[int] 

33 

34 

35class EmcacheClientP(Protocol): 

36 async def add( 

37 self, 

38 key: bytes, 

39 value: bytes, 

40 *, 

41 flags: int = 0, 

42 exptime: int = 0, 

43 noreply: bool = False, 

44 ) -> None: 

45 ... 

46 

47 async def get(self, key: bytes, return_flags: bool = False) -> Optional[ItemP]: 

48 ... 

49 

50 async def gets(self, key: bytes, return_flags: bool = False) -> Optional[ItemP]: 

51 ... 

52 

53 async def increment( 

54 self, key: bytes, value: int, *, noreply: bool = False 

55 ) -> Optional[int]: 

56 ... 

57 

58 async def delete(self, key: bytes, *, noreply: bool = False) -> None: 

59 ... 

60 

61 async def set( 

62 self, 

63 key: bytes, 

64 value: bytes, 

65 *, 

66 flags: int = 0, 

67 exptime: int = 0, 

68 noreply: bool = False, 

69 ) -> None: 

70 ... 

71 

72 async def touch(self, key: bytes, exptime: int, *, noreply: bool = False) -> None: 

73 ... 

74 

75 

76class MemcachedClientP(Protocol): 

77 def add( 

78 self, 

79 key: str, 

80 value: Serializable, 

81 expire: Optional[int] = 0, 

82 noreply: Optional[bool] = None, 

83 flags: Optional[int] = None, 

84 ) -> bool: 

85 ... 

86 

87 def get(self, key: str, default: Optional[str] = None) -> bytes: 

88 ... 

89 

90 def incr(self, key: str, value: int, noreply: Optional[bool] = False) -> int: 

91 ... 

92 

93 def delete(self, key: str, noreply: Optional[bool] = None) -> Optional[bool]: 

94 ... 

95 

96 def set( 

97 self, 

98 key: str, 

99 value: Serializable, 

100 expire: int = 0, 

101 noreply: Optional[bool] = None, 

102 flags: Optional[int] = None, 

103 ) -> bool: 

104 ... 

105 

106 def touch( 

107 self, key: str, expire: Optional[int] = 0, noreply: Optional[bool] = None 

108 ) -> bool: 

109 ... 

110 

111 

112AsyncRedisClient = Union["coredis.Redis[bytes]", "coredis.RedisCluster[bytes]"] 

113RedisClient = Union["redis.Redis", "redis.cluster.RedisCluster"] 

114 

115 

116class ScriptP(Protocol[R_co]): 

117 def __call__(self, keys: List[Serializable], args: List[Serializable]) -> R_co: 

118 ... 

119 

120 

121__all__ = [ 

122 "AsyncRedisClient", 

123 "Callable", 

124 "ClassVar", 

125 "Counter", 

126 "Dict", 

127 "EmcacheClientP", 

128 "ItemP", 

129 "List", 

130 "MemcachedClientP", 

131 "NamedTuple", 

132 "Optional", 

133 "P", 

134 "ParamSpec", 

135 "Protocol", 

136 "ScriptP", 

137 "Serializable", 

138 "TypeVar", 

139 "R", 

140 "R_co", 

141 "RedisClient", 

142 "Tuple", 

143 "Type", 

144 "TypeVar", 

145 "Union", 

146]