I'm using some 3rd-party code which enables locking via routines that are decorated with @contextlib.contextmanager. I'm also using large python3 code base in which we can plug in different locking software, as long as I am able to implement acquire and release methods.
I'm trying to use the 3rd-party code (without knowing how it's written) within this software structure.
To clarify what I'm looking for, suppose that one of the 3rd-party lock routines is written as a standard @contextlib.contextmanager generator, like this:
@contextlib.contextmanager
def lock(arg0, arg1):
try:
# This section of code corresponds to `acquire`.
# Acquire a lock called 'lock', and then ...
yield lock
finally:
# This section of code coresponds to `release`.
# Do cleanup.
It would normally be used like this ...
with third.party.lock(arg0, arg1):
# Do stuff in this critical section
But as I mentioned above, I want to write a class that has an acquire method and a release method which make use of third.party.lock, and I'd like to do it via the existing third.party.lock module, without rewriting it.
In other words, I want to write a class which looks like this ...
class LockWrapper(object):
def __init__(self):
# initialization
def acquire(self):
# Use third.party.lock to obtain a lock.
# ??? ??? ???
def release(self):
# I don't know what to do here. There is no yield
# in the `finally` section of a normal
# @contextlib.contextmanager decorated method/
# ??? ??? ???
As I stated in the comments of my sample code, I don't see how to make acquire and release do anything meaningful.
It looks like I would have to steal code from the original third.party.lock module in order to accomplish this, but I'm hoping that I'm overlooking a way to do this without having to know anything about this 3rd-party code.
Am I out of luck?
Thank you very much.
OK, I figured it out. My answer is based on some of the code here: https://gist.github.com/icio/c0d3f7efd415071f725b
The key is this enter_context helper function, which digs into the structure of the context manager. It's similar to a function by the same name at that site ...
def enter_context(func, *args, **kwargs):
def _acquire():
with func(*args, **kwargs) as f:
yield f
acquire_gen = _acquire()
def release_func():
try:
next(acquire_gen)
except StopIteration:
pass
return acquire_gen, release_func
The following class implements acquire and release for any contextmanager function which is passed to the constructor. I encapsulate enter_context within this class ...
class ContextWrapper(object):
@staticmethod
def enter_context(func, *args, **kwargs):
def _acquire():
with func(*args, **kwargs) as f:
yield f
acquire_gen = _acquire()
def release_func():
try:
next(acquire_gen)
except StopIteration:
pass
return acquire_gen, release_func
def __init__(self, func, *args, **kwargs):
self._acq, self._rel = self.__class__.enter_context(func, *args, **kwargs)
def acquire(self):
next(self._acq)
return True
def release(self):
self._rel()
# Traditional Dijkstra names.
P = acquire
V = release
Then, I can create my wrapper around third.party.lock as follows ...
mylocker = ContextWrapper(third.party.lock, arg0, arg1)
... and I can call acquire and release as follows ...
mylocker.acquire()
# or mylocker.P()
mylocker.release()
# or mylocker.V()
Here's another example of how to use this class ...
class ThirdPartyLockClass(ContextWrapper):
def __init__(self, arg0, arg1):
# do any initialization
super().__init__(third.party.lock, arg0, arg1)
# Implementing the following methods is optional.
# This is only needed if this class intends to do more
# than the wrapped class during `acquire` or `release`,
# such as logging, etc.
def acquire(self):
# do whatever
rc = super().acquire()
# maybe to other stuff
return rc
def release(self):
# do whatever
super().release()
# maybe do other stuff
P = acquire
V = release
mylocker = ThirdPartyLockClass(arg0, arg1)
mylocker.acquire()
# or mylocker.P()
mylocker.release()
# or mylocker.V()
... or I can even do the following in the simple case where no extra functionality is added to the third party lock class ...
class GenericLocker(ContextWrapper):
def __init__(self, func, *args, **kwargs):
super().__init__(func, *args, **kwargs)
mylocker = GenericLocker(third.party.lock, arg0, arg1)
mylocker.acquire()
# or mylocker.P()
mylocker.release()
# or mylocker.V()