1 from types import TracebackType
2 from typing import Any, Text, Type, Union
3 from typing_extensions import Protocol
5 from redis.client import Redis
7 _TokenValue = Union[bytes, Text]
9 class _Local(Protocol):
10 token: _TokenValue | None
18 timeout: float | None = ...,
21 blocking_timeout: float | None = ...,
22 thread_local: bool = ...,
24 def register_scripts(self) -> None: ...
25 def __enter__(self) -> Lock: ...
27 self, exc_type: Type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None
30 self, blocking: bool | None = ..., blocking_timeout: None | int | float = ..., token: _TokenValue | None = ...
32 def do_acquire(self, token: _TokenValue) -> bool: ...
33 def locked(self) -> bool: ...
34 def owned(self) -> bool: ...
35 def release(self) -> None: ...
36 def do_release(self, expected_token: _TokenValue) -> None: ...
37 def extend(self, additional_time: int | float, replace_ttl: bool = ...) -> bool: ...
38 def do_extend(self, additional_time: int | float, replace_ttl: bool) -> bool: ...
39 def reacquire(self) -> bool: ...
40 def do_reacquire(self) -> bool: ...