Passing arguments to context manager

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.

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.