import _collections_abc
from collections import deque
from functools import wraps
+from types import MethodType
__all__ = ["asynccontextmanager", "contextmanager", "closing", "nullcontext",
"AbstractContextManager", "AbstractAsyncContextManager",
@staticmethod
def _create_exit_wrapper(cm, cm_exit):
- def _exit_wrapper(exc_type, exc, tb):
- return cm_exit(cm, exc_type, exc, tb)
- return _exit_wrapper
+ return MethodType(cm_exit, cm)
@staticmethod
def _create_cb_wrapper(callback, *args, **kwds):
def _push_cm_exit(self, cm, cm_exit):
"""Helper to correctly register callbacks to __exit__ methods."""
_exit_wrapper = self._create_exit_wrapper(cm, cm_exit)
- _exit_wrapper.__self__ = cm
self._push_exit_callback(_exit_wrapper, True)
def _push_exit_callback(self, callback, is_sync=True):
@staticmethod
def _create_async_exit_wrapper(cm, cm_exit):
- async def _exit_wrapper(exc_type, exc, tb):
- return await cm_exit(cm, exc_type, exc, tb)
- return _exit_wrapper
+ return MethodType(cm_exit, cm)
@staticmethod
def _create_async_cb_wrapper(callback, *args, **kwds):
"""Helper to correctly register coroutine function to __aexit__
method."""
_exit_wrapper = self._create_async_exit_wrapper(cm, cm_exit)
- _exit_wrapper.__self__ = cm
self._push_exit_callback(_exit_wrapper, False)
async def __aenter__(self):