Back to results for “Threading”
Python › File I/O & Context Managers

How would you implement a context manager for managing a lock (like threading.Lock) while ALSO adding a timeout, raising an error if the lock can't be acquired quickly enough?

Advanced

Answer

Implement __enter__ to attempt lock.acquire(timeout=N), raising a custom TimeoutError if it fails, and __exit__ to always release the lock — combining the standard locking pattern with custom timeout-handling logic that plain 'with lock:' doesn't provide out of the box.
import threading

class TimedLock:
    def __init__(self, lock, timeout):
        self.lock = lock
        self.timeout = timeout
    def __enter__(self):
        if not self.lock.acquire(timeout=self.timeout):
            raise TimeoutError("Could not acquire lock in time")
        return self.lock
    def __exit__(self, *args):
        self.lock.release()

lock = threading.Lock()
with TimedLock(lock, timeout=5):
    critical_section()
Real-world example Preventing a thread from hanging indefinitely waiting for a lock, instead failing fast with a clear timeout error after a reasonable wait.

Common follow-ups: Why does __exit__ need to accept *args (exc_type, exc_value, traceback) even if it doesn't use them?