源碼剖析@contextlib.contextmanager
示例
@contextlib.contextmanager
def result(a):
print('before')
yield
print('after')
外層裝飾源碼
包裝func函數，真實調用func()時，返回的為_GeneratorContextManager對象
def contextmanager(func):
@wraps(func)
def helper(*args, **kwds):
return _GeneratorContextManager(func, args, kwds)
return helper
_GeneratorContextManager對象
該對象實現上下文管理器，繼承父類_GeneratorContextManagerBase，抽象類AbstractContextManager，ContextDecorator
父類_GeneratorContextManagerBase
用來初始化某些_GeneratorContextManager對象需要的屬性，如被包裝的生成器對象，調用生成器時的參數，對象的doc文檔
class _GeneratorContextManagerBase:
"""Shared functionality for @contextmanager and @asynccontextmanager."""
def __init__(self, func, args, kwds):
self.gen = func(*args, **kwds) """保存被裝飾的生成器對象"""
self.func, self.args, self.kwds = func, args, kwds
# Issue 19330: ensure context manager instances have good docstrings
doc = getattr(func, "__doc__", None) """用生成器的doc作為實例的doc，如果沒有，就用類自己的doc作為實例doc"""
if doc is None:
doc = type(self).__doc__
self.__doc__ = doc
# Unfortunately, this still doesn't provide good help output when
# inspecting the created context manager instances, since pydoc
# currently bypasses the instance docstring and shows the docstring
# for the class instead.
# See //bugs.python.org/issue19404 for more details.
抽象類AbstractContextManager
定義類需要實現上下文管理方法
定義判斷是否為AbstractContextManager子類的方法AbstractContextManager，即如果一個類實現了__enter__ and__exit__, 沒有繼承定義判斷是否為AbstractContextManager子類的方法AbstractContextManager，issubclass(classname,AbstractContextManager)也為真
class AbstractContextManager(abc.ABC):
"""An abstract base class for context managers."""
def __enter__(self):
"""Return `self` upon entering the runtime context."""
return self
@abc.abstractmethod
def __exit__(self, exc_type, exc_value, traceback):
"""Raise any exception triggered within the runtime context."""
return None
@classmethod
def __subclasshook__(cls, C):
if cls is AbstractContextManager:
return _collections_abc._check_methods(C, "__enter__", "__exit__")
return NotImplemented
裝飾類ContextDecorator
繼承這個類，可以直接作ContextDecorator對象作為裝飾器，為函數執行提供上下文管理（被contextmanager裝飾的函數，可以作為裝飾器 & with語句使用）
class ContextDecorator(object):
"A base class or mixin that enables context managers to work as decorators."
def _recreate_cm(self):
"""Return a recreated instance of self.
Allows an otherwise one-shot context manager like
_GeneratorContextManager to support use as
a decorator via implicit recreation.
This is a private interface just for _GeneratorContextManager.
See issue #11647 for details.
"""
return self
def __call__(self, func):
@wraps(func)
def inner(*args, **kwds):
with self._recreate_cm():
return func(*args, **kwds)
return inner
示例
class MyContextManager(ContextDecorator):
"Test MyContextManager."
def __enter__(self):
print('enter')
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print('exit')
@MyContextManager()
def report(a):
print(a, 'report function')
return a
執行report(1), 輸出：
eneter
1 report function
exit
1
_GeneratorContextManager類
對象的使用：
@contextlib.contextmanager
def gcm_func(a):
print('before')
print('gcm_func', a)
yield
print('after')
#使用方式1：gcm_func直接作為上下文管理器：
with gcm_func(1):
print('-- in with ---')
輸出：
before
gcm_func 1
-- in with ---
after
#使用方式2： gcm_func作為函數的上下文管理
@gcm_func(1)
def with_func(a):
print('-- in with ---')
with_func(1)
with_func(1)
"""
注意：ContextDecorator中__call__定義了每次調用with_func前，會調用_recreate_cm生成新的_GeneratorContextManager對象作為上下文管理器，所以這邊可以調用2次
否則在第一次with_func(1)就已經清空gcm_func生成器並刪除with_func屬性
"""
輸出：
同方式1
class _GeneratorContextManager(_GeneratorContextManagerBase,
AbstractContextManager,
ContextDecorator):
"""Helper for @contextmanager decorator."""
def _recreate_cm(self):
"""
_GeneratorContextManager實例上下文管理一次就無法再次調用
如果_GeneratorContextManager實例用作裝飾器，每次調用時需要重新生成實例
"""
# _GCM instances are one-shot context managers, so the
# CM must be recreated each time a decorated function is
# called
return self.__class__(self.func, self.args, self.kwds)
def __enter__(self):
"""被裝飾函數的參數只有在初始化實例時有用"""
# do not keep args and kwds alive unnecessarily
# they are only needed for recreation, which is not possible anymore
del self.args, self.kwds, self.func
try:
return next(self.gen)
except StopIteration:
raise RuntimeError("generator didn't yield") from None
def __exit__(self, type, value, traceback):
"""
type: with語句內拋出的異常類
value: with語句內拋出的異常資訊
traceback: with語句內拋出的異常堆棧
"""
a=str(type)
if type is None:
"""
with語句內沒有報錯，往yield停止的部分繼續執行，並會拋出異常
如果往下走沒有遇到stop異常，也就是contextmanager函數有兩個yield，會報錯"""
try:
next(self.gen)
except StopIteration:
return False
else:
raise RuntimeError("generator didn't stop")
else:
"""
如果with中拋出了異常，在yield處拋出異常
"""
if value is None:
# Need to force instantiation so we can reliably
# tell if we get the same exception back
value = type()
try:
self.gen.throw(type, value, traceback)
except StopIteration as exc:
"""如果拋出的異常在yield中被except，不再拋出，而是往下走會拋出生成器的stop異常"""
# Suppress StopIteration *unless* it's the same exception that
# was passed to throw(). This prevents a StopIteration
# raised inside the "with" statement from being suppressed.
return exc is not value
except RuntimeError as exc:
"""如果with中有異常，拋出with中的異常給yield後，如果後續的語句有異常，判斷異常是否為屬於上下文管理器的異常"""
# Don't re-raise the passed in exception.(issue27122)
"""如果是在上下文管理器中except raise異常，不要再拋出"""
if exc is value:
return False
# Likewise, avoid suppressing if a StopIteration exception
# was passed to throw() and later wrapped into a RuntimeError
# (see PEP 479).
"""忽略with中拋出的stop異常，不在這裡拋出異常"""
if type is StopIteration and exc.__cause__ is value:
return False
"""如果是後續語句中其他異常，屬於上下文管理器的異常，拋出"""
raise
except:
# only re-raise if it's *not* the exception that was
# passed to throw(), because __exit__() must not raise
# an exception unless __exit__() itself failed. But throw()
# has to raise the exception to signal propagation, so this
# fixes the impedance mismatch between the throw() protocol
# and the __exit__() protocol.
#
# This cannot use 'except BaseException as exc' (as in the
# async implementation) to maintain compatibility with
# Python 2, where old-style class exceptions are not caught
# by 'except BaseException'.
if sys.exc_info()[1] is value:
return False
raise
"""處理異常之後，往下走還有yield"""
raise RuntimeError("generator didn't stop after throw()")
