Repository: redis/redis-py
Pull request #2137 (2022-04-24 20:49:51)

Made sync lock consistent and added types to it
Fixes: #1348
Needed for: jazzband/django-redis#603

### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [x] Is the new or changed code fully tested?
- [x] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Was the change added to CHANGES file?

### Description of change

Added blocking flag to client.lock and types to sync lock.
Some more nit picking inconsistency was fixed so that regardless on who the lock is used, all the options are the same.

If this is okay I will also add the same to the async version of the lock

codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2137?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
> Merging [#2137](https://codecov.io/gh/redis/redis-py/pull/2137?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (f347e06) into [master](https://codecov.io/gh/redis/redis-py/commit/e5e265de87dfe7ef8cc5cad8c247d067c74fa44d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) (e5e265d) will **decrease** coverage by `9.59%`.
> The diff coverage is `100.00%`.

```diff
@@            Coverage Diff             @@
##           master    #2137      +/-   ##
==========================================
- Coverage   92.44%   82.85%   -9.60%     
==========================================
  Files         104      104              
  Lines       24364    23532     -832     
==========================================
- Hits        22524    19497    -3027     
- Misses       1840     4035    +2195     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2137?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [redis/client.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2xpZW50LnB5) | `88.66% <ø> (-0.81%)` | :arrow_down: |
| [redis/lock.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvbG9jay5weQ==) | `100.00% <100.00%> (ø)` | |
| [redis/typing.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvdHlwaW5nLnB5) | `90.62% <100.00%> (+0.30%)` | :arrow_up: |
| [tests/test\_lock.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9sb2NrLnB5) | `99.50% <100.00%> (-0.50%)` | :arrow_down: |
| [tests/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jbHVzdGVyLnB5) | `14.80% <0.00%> (-82.81%)` | :arrow_down: |
| [redis/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2x1c3Rlci5weQ==) | `22.10% <0.00%> (-69.73%)` | :arrow_down: |
| [redis/crc.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY3JjLnB5) | `40.00% <0.00%> (-60.00%)` | :arrow_down: |
| [redis/commands/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29tbWFuZHMvY2x1c3Rlci5weQ==) | `34.28% <0.00%> (-59.29%)` | :arrow_down: |
| [redis/utils.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvdXRpbHMucHk=) | `71.42% <0.00%> (-17.47%)` | :arrow_down: |
| [redis/exceptions.py](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvZXhjZXB0aW9ucy5weQ==) | `86.56% <0.00%> (-11.95%)` | :arrow_down: |
| ... and [60 more](https://codecov.io/gh/redis/redis-py/pull/2137/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2137?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/redis/redis-py/pull/2137?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Last update [e5e265d...f347e06](https://codecov.io/gh/redis/redis-py/pull/2137?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

WisdomPill: @chayim or @dvora-h friendly ping
dvora-h: Thanks @WisdomPill! It looks great to me! 

As you suggested - I would really appreciate if you add an async version as well
WisdomPill: nice! probably in the weekend I will be back at it :)
WisdomPill: sorry I did not finish working on the async version yet... I just started and got caught up by other things

Patch:
diff --git a/benchmarks/cluster_async.py b/benchmarks/cluster_async.py
index fd2ab460..17dd52b5 100644
--- a/benchmarks/cluster_async.py
+++ b/benchmarks/cluster_async.py
@@ -249,8 +249,8 @@ if __name__ == "__main__":
     port = 16379
     password = None
 
-    count = 1000
-    size = 16
+    count = 10000
+    size = 256
 
     asyncio.run(main("asyncio"))
     asyncio.run(main("asyncio", gather=False))
diff --git a/benchmarks/cluster_async_pipeline.py b/benchmarks/cluster_async_pipeline.py
new file mode 100644
index 00000000..af45b445
--- /dev/null
+++ b/benchmarks/cluster_async_pipeline.py
@@ -0,0 +1,107 @@
+import asyncio
+import functools
+import time
+
+import aioredis_cluster
+import aredis
+import uvloop
+
+import redis.asyncio as redispy
+
+
+def timer(func):
+    @functools.wraps(func)
+    async def wrapper(*args, **kwargs):
+        tic = time.perf_counter()
+        await func(*args, **kwargs)
+        toc = time.perf_counter()
+        return f"{toc - tic:.4f}"
+
+    return wrapper
+
+
+@timer
+async def warmup(client):
+    await asyncio.gather(
+        *(asyncio.create_task(client.exists(f"bench:warmup_{i}")) for i in range(100))
+    )
+
+
+@timer
+async def run(client):
+    data_str = "a" * size
+    data_int = int("1" * size)
+
+    for i in range(count):
+        with client.pipeline() as pipe:
+            await (
+                pipe.set(f"bench:str_{i}", data_str)
+                .set(f"bench:int_{i}", data_int)
+                .get(f"bench:str_{i}")
+                .get(f"bench:int_{i}")
+                .hset("bench:hset", str(i), data_str)
+                .hget("bench:hset", str(i))
+                .incr("bench:incr")
+                .lpush("bench:lpush", data_int)
+                .lrange("bench:lpush", 0, 300)
+                .lpop("bench:lpush")
+                .execute()
+            )
+
+
+async def main(loop):
+    arc = aredis.StrictRedisCluster(
+        host=host,
+        port=port,
+        password=password,
+        max_connections=2**31,
+        max_connections_per_node=2**31,
+        readonly=False,
+        reinitialize_steps=count,
+        skip_full_coverage_check=True,
+        decode_responses=False,
+        max_idle_time=count,
+        idle_check_interval=count,
+    )
+    print(f"{loop} {await warmup(arc)} aredis")
+    print(await run(arc))
+    arc.connection_pool.disconnect()
+
+    aiorc = await aioredis_cluster.create_redis_cluster(
+        [(host, port)],
+        password=password,
+        state_reload_interval=count,
+        idle_connection_timeout=count,
+        pool_maxsize=2**31,
+    )
+    print(f"{loop} {await warmup(aiorc)} aioredis-cluster")
+    print(await run(aiorc))
+    aiorc.close()
+    await aiorc.wait_closed()
+
+    async with redispy.RedisCluster(
+        host=host,
+        port=port,
+        password=password,
+        reinitialize_steps=count,
+        read_from_replicas=False,
+        decode_responses=False,
+        max_connections=2**31,
+    ) as rca:
+        print(f"{loop} {await warmup(rca)} redispy")
+        print(await run(rca))
+
+
+if __name__ == "__main__":
+    host = "localhost"
+    port = 16379
+    password = None
+
+    count = 10000
+    size = 256
+
+    asyncio.run(main("asyncio"))
+
+    uvloop.install()
+
+    asyncio.run(main("uvloop"))
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index 3405a499..b53f8487 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -468,9 +468,8 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         return key_slot(k)
 
     async def _determine_nodes(
-        self, *args: Any, node_flag: Optional[str] = None
+        self, command: str, *args: Any, node_flag: Optional[str] = None
     ) -> List["ClusterNode"]:
-        command = args[0]
         if not node_flag:
             # get the nodes group for this command if it was predefined
             node_flag = self.command_flags.get(command)
@@ -495,16 +494,15 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         # get the node that holds the key's slot
         return [
             self.nodes_manager.get_node_from_slot(
-                await self._determine_slot(*args),
+                await self._determine_slot(command, *args),
                 self.read_from_replicas and command in READ_COMMANDS,
             )
         ]
 
-    async def _determine_slot(self, *args: Any) -> int:
-        command = args[0]
+    async def _determine_slot(self, command: str, *args: Any) -> int:
         if self.command_flags.get(command) == SLOT_ID:
             # The command contains the slot ID
-            return int(args[1])
+            return int(args[0])
 
         # Get the keys in the command
 
@@ -516,19 +514,17 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         # - fix: https://github.com/redis/redis/pull/9733
         if command in ("EVAL", "EVALSHA"):
             # command syntax: EVAL "script body" num_keys ...
-            if len(args) <= 2:
-                raise RedisClusterException(f"Invalid args in command: {args}")
-            num_actual_keys = args[2]
-            eval_keys = args[3 : 3 + num_actual_keys]
+            if len(args) < 2:
+                raise RedisClusterException(
+                    f"Invalid args in command: {command, *args}"
+                )
+            keys = args[2 : 2 + args[1]]
             # if there are 0 keys, that means the script can be run on any node
             # so we can just return a random slot
-            if not eval_keys:
+            if not keys:
                 return random.randrange(0, REDIS_CLUSTER_HASH_SLOTS)
-            keys = eval_keys
         else:
-            keys = await self.commands_parser.get_keys(
-                self.nodes_manager.default_node, *args
-            )
+            keys = await self.commands_parser.get_keys(command, *args)
             if not keys:
                 # FCALL can call a function with 0 keys, that means the function
                 #  can be run on any node so we can just return a random slot
@@ -848,13 +844,13 @@ class ClusterNode:
                 self._free.append(connection)
 
             return self._free.popleft()
-        else:
-            if len(self._connections) < self.max_connections:
-                connection = self.connection_class(**self.connection_kwargs)
-                self._connections.append(connection)
-                return connection
-            else:
-                raise ConnectionError("Too many connections")
+
+        if len(self._connections) < self.max_connections:
+            connection = self.connection_class(**self.connection_kwargs)
+            self._connections.append(connection)
+            return connection
+
+        raise ConnectionError("Too many connections")
 
     async def parse_response(
         self, connection: Connection, command: str, **kwargs: Any
@@ -872,10 +868,10 @@ class ClusterNode:
             raise
 
         # Return response
-        try:
+        if command in self.response_callbacks:
             return self.response_callbacks[command](response, **kwargs)
-        except KeyError:
-            return response
+
+        return response
 
     async def execute_command(self, *args: Any, **kwargs: Any) -> Any:
         # Acquire connection
@@ -891,7 +887,7 @@ class ClusterNode:
             # Release connection
             self._free.append(connection)
 
-    async def execute_pipeline(self) -> None:
+    async def execute_pipeline(self) -> bool:
         # Acquire connection
         connection = self.acquire_connection()
 
@@ -901,17 +897,20 @@ class ClusterNode:
         )
 
         # Read responses
-        try:
-            for cmd in self._command_stack:
-                try:
-                    cmd.result = await self.parse_response(
-                        connection, cmd.args[0], **cmd.kwargs
-                    )
-                except Exception as e:
-                    cmd.result = e
-        finally:
-            # Release connection
-            self._free.append(connection)
+        ret = False
+        for cmd in self._command_stack:
+            try:
+                cmd.result = await self.parse_response(
+                    connection, cmd.args[0], **cmd.kwargs
+                )
+            except Exception as e:
+                cmd.result = e
+                ret = True
+
+        # Release connection
+        self._free.append(connection)
+
+        return ret
 
 
 class NodesManager:
@@ -1257,6 +1256,13 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
     def __await__(self) -> Generator[Any, None, "ClusterPipeline"]:
         return self.initialize().__await__()
 
+    def __enter__(self) -> "ClusterPipeline":
+        self._command_stack = []
+        return self
+
+    def __exit__(self, exc_type: None, exc_value: None, traceback: None) -> None:
+        self._command_stack = []
+
     def __bool__(self) -> bool:
         return bool(self._command_stack)
 
@@ -1310,6 +1316,7 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
 
                 try:
                     return await self._execute(
+                        self._client,
                         self._command_stack,
                         raise_on_error=raise_on_error,
                         allow_redirections=allow_redirections,
@@ -1331,60 +1338,60 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
 
     async def _execute(
         self,
+        client: "RedisCluster",
         stack: List["PipelineCommand"],
         raise_on_error: bool = True,
         allow_redirections: bool = True,
     ) -> List[Any]:
-        client = self._client
+        todo = [
+            cmd for cmd in stack if not cmd.result or isinstance(cmd.result, Exception)
+        ]
+
         nodes = {}
-        for cmd in stack:
-            if not cmd.result or isinstance(cmd.result, Exception):
-                target_nodes = await client._determine_nodes(*cmd.args)
-                if not target_nodes:
-                    raise RedisClusterException(
-                        f"No targets were found to execute {cmd.args} command on"
-                    )
-                if len(target_nodes) > 1:
-                    raise RedisClusterException(
-                        f"Too many targets for command {cmd.args}"
-                    )
+        for cmd in todo:
+            target_nodes = await client._determine_nodes(*cmd.args)
+            if not target_nodes:
+                raise RedisClusterException(
+                    f"No targets were found to execute {cmd.args} command on"
+                )
+            if len(target_nodes) > 1:
+                raise RedisClusterException(f"Too many targets for command {cmd.args}")
 
-                node = target_nodes[0]
-                if node.name not in nodes:
-                    nodes[node.name] = node
-                    node._command_stack = []
-                node._command_stack.append(cmd)
+            node = target_nodes[0]
+            if node.name not in nodes:
+                nodes[node.name] = node
+                node._command_stack = []
+            node._command_stack.append(cmd)
 
-        await asyncio.gather(
+        errors = await asyncio.gather(
             *(asyncio.ensure_future(node.execute_pipeline()) for node in nodes.values())
         )
 
-        if allow_redirections:
-            # send each errored command individually
-            for cmd in stack:
-                if isinstance(cmd.result, (TryAgainError, MovedError, AskError)):
-                    try:
-                        cmd.result = await client.execute_command(
-                            *cmd.args, **cmd.kwargs
+        if any(errors):
+            if allow_redirections:
+                # send each errored command individually
+                for cmd in todo:
+                    if isinstance(cmd.result, (TryAgainError, MovedError, AskError)):
+                        try:
+                            cmd.result = await client.execute_command(
+                                *cmd.args, **cmd.kwargs
+                            )
+                        except Exception as e:
+                            cmd.result = e
+
+            if raise_on_error:
+                for cmd in todo:
+                    result = cmd.result
+                    if isinstance(result, Exception):
+                        command = " ".join(map(safe_str, cmd.args))
+                        msg = (
+                            f"Command # {cmd.position + 1} ({command}) of pipeline "
+                            f"caused error: {result.args}"
                         )
-                    except Exception as e:
-                        cmd.result = e
-
-        responses = [cmd.result for cmd in stack]
-
-        if raise_on_error:
-            for cmd in stack:
-                result = cmd.result
-                if isinstance(result, Exception):
-                    command = " ".join(map(safe_str, cmd.args))
-                    msg = (
-                        f"Command # {cmd.position + 1} ({command}) of pipeline "
-                        f"caused error: {result.args}"
-                    )
-                    result.args = (msg,) + result.args[1:]
-                    raise result
+                        result.args = (msg,) + result.args[1:]
+                        raise result
 
-        return responses
+        return [cmd.result for cmd in stack]
 
     def _split_command_across_slots(
         self, command: str, *keys: KeyT
diff --git a/redis/asyncio/parser.py b/redis/asyncio/parser.py
index 62863513..9518c3f1 100644
--- a/redis/asyncio/parser.py
+++ b/redis/asyncio/parser.py
@@ -22,13 +22,16 @@ class CommandsParser:
     So, don't use this with EVAL or EVALSHA.
     """
 
-    __slots__ = ("commands",)
+    __slots__ = ("commands", "node")
 
     def __init__(self) -> None:
         self.commands: Dict[str, Union[int, Dict[str, Any]]] = {}
 
-    async def initialize(self, r: "ClusterNode") -> None:
-        commands = await r.execute_command("COMMAND")
+    async def initialize(self, node: Optional["ClusterNode"] = None) -> None:
+        if node:
+            self.node = node
+
+        commands = await self.node.execute_command("COMMAND")
         for cmd, command in commands.items():
             if "movablekeys" in command["flags"]:
                 commands[cmd] = -1
@@ -41,9 +44,7 @@ class CommandsParser:
     # As soon as this PR is merged into Redis, we should reimplement
     # our logic to use COMMAND INFO changes to determine the key positions
     # https://github.com/redis/redis/pull/8324
-    async def get_keys(
-        self, redis_conn: "ClusterNode", *args: Any
-    ) -> Optional[Tuple[str, ...]]:
+    async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
         if len(args) < 2:
             # The command has no keys in it
             return None
@@ -58,7 +59,7 @@ class CommandsParser:
             if cmd_name not in self.commands:
                 # We'll try to reinitialize the commands cache, if the engine
                 # version has changed, the commands may not be current
-                await self.initialize(redis_conn)
+                await self.initialize()
                 if cmd_name not in self.commands:
                     raise RedisError(
                         f"{cmd_name.upper()} command doesn't exist in Redis commands"
@@ -71,18 +72,16 @@ class CommandsParser:
         if command == 0:
             return None
         if command == -1:
-            return await self._get_moveable_keys(redis_conn, *args)
+            return await self._get_moveable_keys(*args)
 
         last_key_pos = command["last_key_pos"]
         if last_key_pos < 0:
             last_key_pos = len(args) + last_key_pos
         return args[command["first_key_pos"] : last_key_pos + 1 : command["step_count"]]
 
-    async def _get_moveable_keys(
-        self, redis_conn: "ClusterNode", *args: Any
-    ) -> Optional[Tuple[str, ...]]:
+    async def _get_moveable_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
         try:
-            keys = await redis_conn.execute_command("COMMAND GETKEYS", *args)
+            keys = await self.node.execute_command("COMMAND GETKEYS", *args)
         except ResponseError as e:
             message = e.__str__()
             if (
diff --git a/redis/client.py b/redis/client.py
index 58668eeb..fcc2758d 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -1098,6 +1098,7 @@ class Redis(AbstractRedis, RedisModuleCommands, CoreCommands, SentinelCommands):
         name,
         timeout=None,
         sleep=0.1,
+        blocking=True,
         blocking_timeout=None,
         lock_class=None,
         thread_local=True,
@@ -1113,6 +1114,12 @@ class Redis(AbstractRedis, RedisModuleCommands, CoreCommands, SentinelCommands):
         when the lock is in blocking mode and another client is currently
         holding the lock.
 
+        ``blocking`` indicates whether calling ``acquire`` should block until
+        the lock has been acquired or to fail immediately, causing ``acquire``
+        to return False and the lock not being acquired. Defaults to True.
+        Note this value can be overridden by passing a ``blocking``
+        argument to ``acquire``.
+
         ``blocking_timeout`` indicates the maximum amount of time in seconds to
         spend trying to acquire the lock. A value of ``None`` indicates
         continue trying forever. ``blocking_timeout`` can be specified as a
@@ -1155,6 +1162,7 @@ class Redis(AbstractRedis, RedisModuleCommands, CoreCommands, SentinelCommands):
             name,
             timeout=timeout,
             sleep=sleep,
+            blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
         )
diff --git a/redis/cluster.py b/redis/cluster.py
index a88792bc..8e4c654c 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -766,6 +766,7 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
         name,
         timeout=None,
         sleep=0.1,
+        blocking=True,
         blocking_timeout=None,
         lock_class=None,
         thread_local=True,
@@ -781,6 +782,12 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
         when the lock is in blocking mode and another client is currently
         holding the lock.
 
+        ``blocking`` indicates whether calling ``acquire`` should block until
+        the lock has been acquired or to fail immediately, causing ``acquire``
+        to return False and the lock not being acquired. Defaults to True.
+        Note this value can be overridden by passing a ``blocking``
+        argument to ``acquire``.
+
         ``blocking_timeout`` indicates the maximum amount of time in seconds to
         spend trying to acquire the lock. A value of ``None`` indicates
         continue trying forever. ``blocking_timeout`` can be specified as a
@@ -823,6 +830,7 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
             name,
             timeout=timeout,
             sleep=sleep,
+            blocking=blocking,
             blocking_timeout=blocking_timeout,
             thread_local=thread_local,
         )
diff --git a/redis/lock.py b/redis/lock.py
index 74e769bf..c509f7d9 100644
--- a/redis/lock.py
+++ b/redis/lock.py
@@ -1,9 +1,11 @@
 import threading
 import time as mod_time
 import uuid
-from types import SimpleNamespace
+from types import SimpleNamespace, TracebackType
+from typing import Optional, Type
 
 from redis.exceptions import LockError, LockNotOwnedError
+from redis.typing import Number
 
 
 class Lock:
@@ -74,12 +76,13 @@ class Lock:
     def __init__(
         self,
         redis,
-        name,
-        timeout=None,
-        sleep=0.1,
-        blocking=True,
-        blocking_timeout=None,
-        thread_local=True,
+        name: str,
+        *,
+        timeout: Optional[Number] = None,
+        sleep: Number = 0.1,
+        blocking: bool = True,
+        blocking_timeout: Optional[Number] = None,
+        thread_local: bool = True,
     ):
         """
         Create a new Lock instance named ``name`` using the Redis client
@@ -142,7 +145,7 @@ class Lock:
         self.local.token = None
         self.register_scripts()
 
-    def register_scripts(self):
+    def register_scripts(self) -> None:
         cls = self.__class__
         client = self.redis
         if cls.lua_release is None:
@@ -152,15 +155,27 @@ class Lock:
         if cls.lua_reacquire is None:
             cls.lua_reacquire = client.register_script(cls.LUA_REACQUIRE_SCRIPT)
 
-    def __enter__(self):
+    def __enter__(self) -> "Lock":
         if self.acquire():
             return self
         raise LockError("Unable to acquire lock within the time specified")
 
-    def __exit__(self, exc_type, exc_value, traceback):
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_value: Optional[BaseException],
+        traceback: Optional[TracebackType],
+    ) -> None:
         self.release()
 
-    def acquire(self, blocking=None, blocking_timeout=None, token=None):
+    def acquire(
+        self,
+        *,
+        sleep: Optional[Number] = None,
+        blocking: Optional[bool] = None,
+        blocking_timeout: Optional[Number] = None,
+        token: Optional[str] = None,
+    ):
         """
         Use Redis to hold a shared, distributed lock named ``name``.
         Returns True once the lock is acquired.
@@ -176,7 +191,8 @@ class Lock:
         object with the default encoding. If a token isn't specified, a UUID
         will be generated.
         """
-        sleep = self.sleep
+        if sleep is None:
+            sleep = self.sleep
         if token is None:
             token = uuid.uuid1().hex.encode()
         else:
@@ -200,7 +216,7 @@ class Lock:
                 return False
             mod_time.sleep(sleep)
 
-    def do_acquire(self, token):
+    def do_acquire(self, token: str) -> bool:
         if self.timeout:
             # convert to milliseconds
             timeout = int(self.timeout * 1000)
@@ -210,13 +226,13 @@ class Lock:
             return True
         return False
 
-    def locked(self):
+    def locked(self) -> bool:
         """
         Returns True if this key is locked by any process, otherwise False.
         """
         return self.redis.get(self.name) is not None
 
-    def owned(self):
+    def owned(self) -> bool:
         """
         Returns True if this key is locked by this lock, otherwise False.
         """
@@ -228,21 +244,23 @@ class Lock:
             stored_token = encoder.encode(stored_token)
         return self.local.token is not None and stored_token == self.local.token
 
-    def release(self):
-        "Releases the already acquired lock"
+    def release(self) -> None:
+        """
+        Releases the already acquired lock
+        """
         expected_token = self.local.token
         if expected_token is None:
             raise LockError("Cannot release an unlocked lock")
         self.local.token = None
         self.do_release(expected_token)
 
-    def do_release(self, expected_token):
+    def do_release(self, expected_token: str) -> None:
         if not bool(
             self.lua_release(keys=[self.name], args=[expected_token], client=self.redis)
         ):
             raise LockNotOwnedError("Cannot release a lock" " that's no longer owned")
 
-    def extend(self, additional_time, replace_ttl=False):
+    def extend(self, additional_time: int, replace_ttl: bool = False) -> bool:
         """
         Adds more time to an already acquired lock.
 
@@ -259,19 +277,19 @@ class Lock:
             raise LockError("Cannot extend a lock with no timeout")
         return self.do_extend(additional_time, replace_ttl)
 
-    def do_extend(self, additional_time, replace_ttl):
+    def do_extend(self, additional_time: int, replace_ttl: bool) -> bool:
         additional_time = int(additional_time * 1000)
         if not bool(
             self.lua_extend(
                 keys=[self.name],
-                args=[self.local.token, additional_time, replace_ttl and "1" or "0"],
+                args=[self.local.token, additional_time, "1" if replace_ttl else "0"],
                 client=self.redis,
             )
         ):
-            raise LockNotOwnedError("Cannot extend a lock that's" " no longer owned")
+            raise LockNotOwnedError("Cannot extend a lock that's no longer owned")
         return True
 
-    def reacquire(self):
+    def reacquire(self) -> bool:
         """
         Resets a TTL of an already acquired lock back to a timeout value.
         """
@@ -281,12 +299,12 @@ class Lock:
             raise LockError("Cannot reacquire a lock with no timeout")
         return self.do_reacquire()
 
-    def do_reacquire(self):
+    def do_reacquire(self) -> bool:
         timeout = int(self.timeout * 1000)
         if not bool(
             self.lua_reacquire(
                 keys=[self.name], args=[self.local.token, timeout], client=self.redis
             )
         ):
-            raise LockNotOwnedError("Cannot reacquire a lock that's" " no longer owned")
+            raise LockNotOwnedError("Cannot reacquire a lock that's no longer owned")
         return True
diff --git a/redis/typing.py b/redis/typing.py
index 6748612f..b572b0cf 100644
--- a/redis/typing.py
+++ b/redis/typing.py
@@ -11,6 +11,7 @@ if TYPE_CHECKING:
     from redis.connection import ConnectionPool, Encoder
 
 
+Number = Union[int, float]
 EncodedT = Union[bytes, memoryview]
 DecodedT = Union[str, int, float]
 EncodableT = Union[EncodedT, DecodedT]


Test patch:
diff --git a/tests/test_lock.py b/tests/test_lock.py
index 0a63f1e0..10ad7e15 100644
--- a/tests/test_lock.py
+++ b/tests/test_lock.py
@@ -116,6 +116,16 @@ class TestLock:
             assert r.get("foo") == lock.local.token
         assert r.get("foo") is None
 
+    def test_context_manager_blocking_timeout(self, r):
+        with self.get_lock(r, "foo", blocking=False):
+            bt = 0.4
+            sleep = 0.05
+            lock2 = self.get_lock(r, "foo", sleep=sleep, blocking_timeout=bt)
+            start = time.monotonic()
+            assert not lock2.acquire()
+            # The elapsed duration should be less than the total blocking_timeout
+            assert bt > (time.monotonic() - start) > bt - sleep
+
     def test_context_manager_raises_when_locked_not_acquired(self, r):
         r.set("foo", "bar")
         with pytest.raises(LockError):
@@ -221,6 +231,16 @@ class TestLock:
         with pytest.raises(LockNotOwnedError):
             lock.reacquire()
 
+    def test_context_manager_reacquiring_lock_with_no_timeout_raises_error(self, r):
+        with self.get_lock(r, "foo", timeout=None, blocking=False) as lock:
+            with pytest.raises(LockError):
+                lock.reacquire()
+
+    def test_context_manager_reacquiring_lock_no_longer_owned_raises_error(self, r):
+        with pytest.raises(LockNotOwnedError):
+            with self.get_lock(r, "foo", timeout=10, blocking=False):
+                r.set("foo", "a")
+
 
 class TestLockClassSelection:
     def test_lock_class_argument(self, r):