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

1from typing import ( 

2 TYPE_CHECKING, 

3 Any, 

4 Awaitable, 

5 Callable, 

6 Dict, 

7 List, 

8 NamedTuple, 

9 Optional, 

10 Tuple, 

11 Type, 

12 TypeVar, 

13 Union, 

14) 

15 

16from typing_extensions import ClassVar, Counter, ParamSpec, Protocol, TypeAlias 

17 

18Serializable = Union[int, str, float] 

19 

20R = TypeVar("R") 

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

22P = ParamSpec("P") 

23 

24 

25if TYPE_CHECKING: 

26 import coredis 

27 import coredis.commands.script 

28 import pymongo 

29 import redis 

30 

31 

32class ItemP(Protocol): 

33 value: bytes 

34 flags: Optional[int] 

35 cas: Optional[int] 

36 

37 

38class EmcacheClientP(Protocol): 

39 async def add( 

40 self, 

41 key: bytes, 

42 value: bytes, 

43 *, 

44 flags: int = 0, 

45 exptime: int = 0, 

46 noreply: bool = False, 

47 ) -> None: ... 

48 

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

50 

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

52 

53 async def increment( 

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

55 ) -> Optional[int]: ... 

56 

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

58 

59 async def set( 

60 self, 

61 key: bytes, 

62 value: bytes, 

63 *, 

64 flags: int = 0, 

65 exptime: int = 0, 

66 noreply: bool = False, 

67 ) -> None: ... 

68 

69 async def touch( 

70 self, key: bytes, exptime: int, *, noreply: bool = False 

71 ) -> None: ... 

72 

73 

74class MemcachedClientP(Protocol): 

75 def add( 

76 self, 

77 key: str, 

78 value: Serializable, 

79 expire: Optional[int] = 0, 

80 noreply: Optional[bool] = None, 

81 flags: Optional[int] = None, 

82 ) -> bool: ... 

83 

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

85 

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

87 

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

89 

90 def set( 

91 self, 

92 key: str, 

93 value: Serializable, 

94 expire: int = 0, 

95 noreply: Optional[bool] = None, 

96 flags: Optional[int] = None, 

97 ) -> bool: ... 

98 

99 def touch( 

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

101 ) -> bool: ... 

102 

103 

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

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

106 

107 

108class ScriptP(Protocol[R_co]): 

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

110 

111 

112MongoClient: TypeAlias = "pymongo.MongoClient[Dict[str, Any]]" # type:ignore[misc] 

113MongoDatabase: TypeAlias = "pymongo.database.Database[Dict[str, Any]]" # type:ignore[misc] 

114MongoCollection: TypeAlias = "pymongo.collection.Collection[Dict[str, Any]]" # type:ignore[misc] 

115 

116__all__ = [ 

117 "AsyncRedisClient", 

118 "Awaitable", 

119 "Callable", 

120 "ClassVar", 

121 "Counter", 

122 "Dict", 

123 "EmcacheClientP", 

124 "ItemP", 

125 "List", 

126 "MemcachedClientP", 

127 "MongoClient", 

128 "MongoCollection", 

129 "MongoDatabase", 

130 "NamedTuple", 

131 "Optional", 

132 "P", 

133 "ParamSpec", 

134 "Protocol", 

135 "ScriptP", 

136 "Serializable", 

137 "TypeVar", 

138 "R", 

139 "R_co", 

140 "RedisClient", 

141 "Tuple", 

142 "Type", 

143 "TypeVar", 

144 "Union", 

145]