Python Redis Lock

4.0.1 · active · verified Sun Apr 12

python-redis-lock is a Python library that provides a distributed lock context manager, implemented using Redis's `SETNX` (SET if Not eXists) and `BLPOP` operations. It aims to offer an interface similar to Python's built-in `threading.Lock`. The current version is 4.0.1, and it maintains an active release cadence, with its latest major update in late 2022.

Warnings

breaking Version 4.0.0 dropped support for Python 2.7 and Python 3.6. Users on these older Python versions must use `python-redis-lock < 4.0.0`.
Fix: Upgrade to Python 3.7+ or pin `python-redis-lock` to a version prior to 4.0.0.
gotcha If an application crashes while holding a lock and `expire` (timeout) is not set, the lock may remain in Redis indefinitely, causing permanent deadlocks. Setting `expire` alone will release the lock after a fixed duration, but if the critical section takes longer, the lock might be released prematurely.
Fix: Always use the `auto_renewal=True` parameter in conjunction with `expire` when creating a `Lock` instance. This ensures the lock is automatically renewed as long as the Python process is running within the `with` block, and will expire cleanly if the process terminates unexpectedly. Example: `Lock(client, 'my-lock', expire=60, auto_renewal=True)`.
deprecated The `force` parameter on the `release()` method was removed in version 3.0.0. This option was considered to encourage sloppy programming and is no longer available.
Fix: Review your code and remove any usage of `lock.release(force=True)`. Ensure proper lock acquisition and release logic without relying on force-releasing locks, which can lead to race conditions.
gotcha `python-redis-lock` implements a distributed lock based on Redis's `SETNX` and `BLPOP`. It is *not* an implementation of the more complex Redlock algorithm (which involves multiple independent Redis instances for higher fault tolerance).
Fix: Understand that this library is suitable for single-instance Redis distributed locking. If your application requires the guarantees of the Redlock algorithm (e.g., across multiple Redis masters), you will need to use a different library (e.g., `redlock-py`) or implement Redlock logic yourself. Do not confuse `redis_lock.Lock` with `redis.lock.Lock` (from `redis-py`), as they are separate implementations.

Install

pip install python-redis-lock Install stable version

Imports

Lock
```
from redis_lock import Lock
```
The `redis` (redis-py) library also has a `Lock` class under `redis.lock`. Ensure you import from `redis_lock` for this specific library's implementation.

Quickstart

This quickstart demonstrates how to acquire and use a distributed lock with `python-redis-lock` using a context manager. It highlights setting an expiration (`expire`) and enabling automatic renewal (`auto_renewal`) to prevent locks from being held indefinitely if the application crashes. It also shows checking the lock status.

import redis
import time
import os
from redis_lock import Lock

REDIS_HOST = os.environ.get('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.environ.get('REDIS_PORT', 6379))

# Connect to Redis
client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)

lock_name = "my-distributed-lock"

# Acquire a lock using a context manager, with a 10-second expiration
# and automatic renewal while the 'with' block is active.
print(f"Attempting to acquire lock '{lock_name}'...")
with Lock(client, lock_name, expire=10, auto_renewal=True, blocking_timeout=5) as lock:
    if lock.acquired:
        print(f"Lock '{lock_name}' acquired! Doing some critical work...")
        # Simulate work that takes longer than the expire time
        # auto_renewal will keep the lock alive.
        time.sleep(15)
        print(f"Work finished. Lock '{lock_name}' will be automatically released.")
    else:
        print(f"Failed to acquire lock '{lock_name}'. Another process holds it.")

# You can also check if a lock is currently held (by anyone)
if Lock(client, lock_name).locked():
    print(f"Lock '{lock_name}' is currently held by someone else (or was not properly released).")
else:
    print(f"Lock '{lock_name}' is not currently held.")

client.close()

view raw JSON →