VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2025-12-15 16:27:08 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

[redis] add overload for blpop and brpop timeout (#4998)

Browse Source 
0 is the default value for the timeout argument for both blpop and brpop. When the timeout is `0`, the return type is non-nullable. Otherwise the return type is optional.

I tested my change with the following code
```python
from typing import Optional, Tuple
import redis

def test_blpop_timeout(r: redis.Redis) -> None:
    a: Tuple[bytes, bytes] = r.blpop('')
    b: Tuple[bytes, bytes] = r.blpop('',timeout=0)
    c: Optional[Tuple[bytes, bytes]] = r.blpop('', timeout=1)
    d: Optional[Tuple[bytes, bytes]] = r.blpop('', timeout=1.0)

def test_brpop_timeout(r: redis.Redis) -> None:
    a: Tuple[bytes, bytes] = r.brpop('')
    b: Tuple[bytes, bytes] = r.brpop('',timeout=0)
    c: Optional[Tuple[bytes, bytes]] = r.brpop('', timeout=1)
    d: Optional[Tuple[bytes, bytes]] = r.brpop('', timeout=1.0)
```
This commit is contained in:
Christopher Dignam
2021-02-04 23:32:34 -05:00
committed by GitHub
parent 2682c1d00a
commit 93e2806232
1 changed files with 8 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
stubs/redis/redis/client.pyi
View File

@@ -473,8 +473,14 @@ class Redis(Generic[_StrType]):
def watch(self, *names): ...
def unlink(self, *names: _Key) -> int: ...
def unwatch(self): ...
def blpop(self, keys: Union[_Value, Iterable[_Value]], timeout: float = ...) -> Optional[Tuple[_StrType, _StrType]]: ...
def brpop(self, keys: Union[_Value, Iterable[_Value]], timeout: float = ...) -> Optional[Tuple[_StrType, _StrType]]: ...
@overload
def blpop(self, keys: Union[_Value, Iterable[_Value]], timeout: Literal[0] = ...) -> Tuple[_StrType, _StrType]: ...
@overload
def blpop(self, keys: Union[_Value, Iterable[_Value]], timeout: float) -> Optional[Tuple[_StrType, _StrType]]: ...
@overload
def brpop(self, keys: Union[_Value, Iterable[_Value]], timeout: Literal[0] = ...) -> Tuple[_StrType, _StrType]: ...
@overload
def brpop(self, keys: Union[_Value, Iterable[_Value]], timeout: float) -> Optional[Tuple[_StrType, _StrType]]: ...
def brpoplpush(self, src, dst, timeout=...): ...
def lindex(self, name: _Key, index: int) -> Optional[_StrType]: ...
def linsert(