Retrieve a redis-py Lock name from a LockError?

Question

I have an application that uses redis.lock.Lock (from the official redis-py package) a lot. It's either used directly by what I'm developing, or by other packages. In rare cases, I end up with LockError (in case the application was restarted incorrectly, for example).

I manage to catch these Exceptions, but in the available attributes, there's only the error message, and not the name of the Lock (which would allow me to do a cleanup task).

Is there any way to add information to the LockError object, or how to find the Lock name from the within except block (bearing in mind that I can't modify the Lock implementation, or make a class that inherits it, as it's might be instantiated from code I don't own)?

Example of code:

try:
  do_many_things_with_many_locks()
except LockError as le:
  # How to retrieve which lock failed?

Answer 1

Given that you can't subclass and modify the Lock implementation, you can introspect the exception and retrieve name from the Lock object.

import inspect

def getlockname(error):
    # This is the frame before the error was raised.
    frame_info = inspect.trace(context=1).pop()
    # We can get Lock instance in the locals in that frame
    lock = inspect.getargvalues(frame_info.frame).locals['self']
    return lock.name

For example,

class LError(Exception):
    pass

class LOErrorWhenAcquired:
    def __init__(self):
        self.name = "foo"
      
    def __enter__(self):
        if self.acquire():
            return self
        raise LError("failed")
  
    def acquire(self):
        return False

    def release(self):
        pass
    
    def __exit__(self):
        pass
    
class LOErrorWhenReleased:
    def __init__(self):
        self.name = "foo"
      
    def __enter__(self):
        if self.acquire():
            return self
      
    def acquire(self):
        return True
  
    def release(self):
        raise LError("failed")

    def __exit__(self, *args):
        self.release()

try:
    with LOErrorWhenAcquired() as wa:
        pass
except LError as e:
    print(getlockname(e))


try:
    with LOErrorWhenReleased() as wr:
        pass
except LError as e:
    print(getlockname(e))