Is there a way to pass arguments using context manager? Here is what I’m trying to do:
async with self.request.app['expiration_lock']('param'):
But I am getting an error:
TypeError: 'OrderStatusLock' object is not callable
Class OrderStatusLock:
class OrderStatusLock(_ContextManagerMixin, Lock): def __init__(self, *args, loop=None): print(args) self._waiters = None self._locked = False if loop is None: self._loop = events.get_event_loop() else: self._loop = loop async def acquire(self, *args): print('acq', args) if (not self._locked and (self._waiters is None or all(w.cancelled() for w in self._waiters))): self._locked = True return True if self._waiters is None: self._waiters = collections.deque() fut = self._loop.create_future() self._waiters.append(fut) try: try: await fut finally: self._waiters.remove(fut) except CancelledError: if not self._locked: self._wake_up_first() raise self._locked = True return True
And if it is possible, what issues I can face, using this? Thank you very much.
Advertisement
Answer
There’s a lot going on in your question, and I don’t know where your _ContextManagerMixin
class comes from. I also don’t know much about async.
However, here’s a simple (non-async) demonstration of a pattern where an argument can be passed to a context manager that alters how the __enter__
method of the context manager operates.
Remember: a context manager is, at its heart, just a class that implements an __enter__
method and an __exit__
method. The __enter__
method is called at the start of the with
block, and the __exit__
method is called at the end of the with
block.
The __call__
method added here in my example class is called immediately before the __enter__
method and, unlike the __enter__
method, can be called with arguments. The __exit__
method takes care to clean up the changes made to the class’s internal state by the __call__
method and the __enter__
method.
from threading import Lock class LockWrapper: def __init__(self): self.lock = Lock() self.food = '' def __enter__(self): self.lock.acquire() print(f'{self.food} for breakfast, please!') return self def __exit__(self, exc_type, exc_val, exc_tb): self.lock.release() self.food = '' return True def __call__(self, spam_preferred: bool): if spam_preferred: self.food = 'Spam' else: self.food = 'Eggs' return self breakfast_context_manager = LockWrapper() with breakfast_context_manager(spam_preferred=True): # prints 'Spam for breakfast, please!' pass with breakfast_context_manager(spam_preferred=False): # prints 'Eggs for breakfast, please!' pass
N.B. My example above will suppress all exceptions that are raised in the body of the with
statement. If you don’t want to suppress any exceptions, or if you only want to suppress certain kinds of exceptions, you’ll need to alter the implementation of the __exit__
method.
Note also that the return values of these functions are quite important. __call__
has to return self
if you want the class to be able to then call __enter__
. __enter__
has to return self
if you want to be able to access the context manager’s internal state in the body of the with
statement. __exit__
should return True
if you want exceptions to be suppressed, and False
if you want an encountered exception to continue to endure outside of the with
statement.
You can find a good tutorial on context managers here. The tutorial also contains some info on how you might adapt the above pattern for async code using the __aenter__
and __aexit__
methods.