Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/limits/typing.py: 3%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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)
16from typing_extensions import ClassVar, Counter, ParamSpec, Protocol, TypeAlias
18Serializable = Union[int, str, float]
20R = TypeVar("R")
21R_co = TypeVar("R_co", covariant=True)
22P = ParamSpec("P")
25if TYPE_CHECKING:
26 import coredis
27 import coredis.commands.script
28 import pymongo
29 import redis
32class ItemP(Protocol):
33 value: bytes
34 flags: Optional[int]
35 cas: Optional[int]
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: ...
49 async def get(self, key: bytes, return_flags: bool = False) -> Optional[ItemP]: ...
51 async def gets(self, key: bytes, return_flags: bool = False) -> Optional[ItemP]: ...
53 async def increment(
54 self, key: bytes, value: int, *, noreply: bool = False
55 ) -> Optional[int]: ...
57 async def delete(self, key: bytes, *, noreply: bool = False) -> None: ...
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: ...
69 async def touch(
70 self, key: bytes, exptime: int, *, noreply: bool = False
71 ) -> None: ...
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: ...
84 def get(self, key: str, default: Optional[str] = None) -> bytes: ...
86 def incr(self, key: str, value: int, noreply: Optional[bool] = False) -> int: ...
88 def delete(self, key: str, noreply: Optional[bool] = None) -> Optional[bool]: ...
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: ...
99 def touch(
100 self, key: str, expire: Optional[int] = 0, noreply: Optional[bool] = None
101 ) -> bool: ...
104AsyncRedisClient = Union["coredis.Redis[bytes]", "coredis.RedisCluster[bytes]"]
105RedisClient = Union["redis.Redis[bytes]", "redis.cluster.RedisCluster[bytes]"]
108class ScriptP(Protocol[R_co]):
109 def __call__(self, keys: List[Serializable], args: List[Serializable]) -> R_co: ...
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]
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]