Python 学习笔记——枪版 contextlib.contextmanager

xhz123456789 · 2026-05-20 06:22:42 · 科技·工程

本文将介绍 contextlib.contextmanager，带你实现一个民间版的 contextlib.contextmanager，并在过程中详细介绍装饰器、生成器、上下文管理器。

:::info[前置知识]{open} python 基础、面向对象、泛型。 :::

:::info[环境]{open} 本文在 fedora43 上使用 python3.14 测试。 :::

:::info[AI 使用说明]{open} 本文在创作过程中使用了生成式 AI

使用 Gemini 用于查找部分资料、检查错误、以及一些润色建议。 :::

0. 前言

最近了解了 python 的上下文管理器，很多教程推荐使用 contextlib.contextmanager 装饰器配合生成器函数来写，正好我们自己实现一个。

1

先想想我们需要什么吧

像这样写一个生成器函数，并且用 @contextmanager 装饰。

:::warning[一些细节]{open} 本文为了方便，不对函数（function）、方法（method）和可调用对象（Callable）做严谨的区分，在本文的场景中，这不太重要。

除非特别说明，可以认为本文中它们都指 typing.Callable，即一切可以通过 obj() 语法使用的对象。 :::

from contextlib import contextmanager

@contextmanager
def managed_resource(*args, **kwds):
    # Code to acquire resource, e.g.:
    resource = acquire_resource(*args, **kwds)
    try:
        yield resource
    finally:
        # Code to release resource, e.g.:
        release_resource(resource)

然后就可以把它当成上下文管理器来用。

with managed_resource(timeout=3600) as resource:
    # Resource is released at the end of this block,
    # even if code in the block raises an exception

基本上，yield 前面的内容会在进入时执行，并且值会给到 as 后面的内容。最后，退出时，函数会执行到结束。

所以，我们需要写一个转换工具，把一个生成器函数包装成上下文管理器，并且在正确的时机调用它。

那么，你大概要问了：

等等，问题是，怎么包装它？？？

好问题，先想想我们拿到了什么吧？

毫无疑问，一个生成器函数。

那么，我们需要拿出什么东西呢？

当然是一个上下文管理器了。

那么，我们就要在正确的时机调用它了。

这才是最大的问题啊……

让我们回忆一下最早的两个问题，想想具体给我们什么？又具体需要提供什么？

生成器函数和上下文管理器，这两个答案，足够你解决问题吗？

显然不够，那么你就需要更细的思考它们。也就是说，它们又分别是什么？或者说，上下文管理器怎么知道 with 走到哪了，我们怎么控制生成器的执行？

生成器（提示 1）

生成器函数，一种调用了返回一个生成器的函数，你只要在函数里面写 yield 它就是生成器函数。

根据 typing.Generator 的定义，生成器是 type((lambda: (yield))())。

就像废话一样。

:::info[不要点开] Gemini 曾建议我移除前面两行内容，理由是可以用 types.GeneratorType， 但这个也是废话。

    def _g():
       yield 1
   GeneratorType = type(_g())

:::

让我们看看交互式环境给我们什么吧：

>>> g=type((lambda: (yield))())
>>> g
<class 'generator'>
>>> dir(g)
['__class__', '__class_getitem__', '__del__', '__delattr__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__getstate__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__iter__', '__le__', '__lt__', '__name__', '__ne__', '__new__', '__next__', '__qualname__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', 'close', 'gi_code', 'gi_frame', 'gi_running', 'gi_suspended', 'gi_yieldfrom', 'send', 'throw']
>>> 

python 说，这就是一个生成器。

我们重点关注 __next__ 方法。

>>> help(g.__next__)

Help on method descriptor __next__:

__next__(self, /) unbound builtins.generator method
    Implement next(self).

__next__ 用来往后获取一个 yeild 的值，通常情况下，你应该用 next(obj) 来调用它。

函数结束时抛出 StopIteration 异常。

上下文管理器 （提示 1）

常见的语法是这样的：

with context_manager as x:
    ...

根据 typing.ContextManager 的说法：

实际上这被链接到了 contextlib.AbstractContextManager。

class AbstractContextManager(abc.ABC):

    """An abstract base class for context managers."""

    __class_getitem__ = classmethod(GenericAlias)

    __slots__ = ()

    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

基本上，一个对象有 __enter__ 和 __exit__ 就能被称为上下文管理器了。

__enter__() 会在进入 with 块的时候调用，其返回值会赋值给 as 后的名称（如果有 as 的话）。

__exit__(self, exc_type, exc_value, traceback) 方法在离开 with 块的时候调用。

开始写吧

现在，再想想那个问题，该怎么在正确的时机调用它呢？是不是已经有答案了？

我们只要让新的东西在被调用的时候返回一个包装过的类。

在进入的时候，也就是 __enter__()，用 next 让函数执行到第一个 yield 处，并且把这个值返回。

在退出的时候，用 next 让函数执行到结尾然后捕获 StopIteration 。

def contextmanager(func):
    class Wrapper:
        def __init__(self,*args,**kwargs):
            self.g=func(*args,**kwargs)
        def __enter__(self):
            return next(self.g)
        def __exit__(self, exc_type, exc_value, traceback):
            try:
                next(self.g)
            except StopIteration:
                pass
    return Wrapper

2

这似乎的确能用了，不过，用起来好像有点难受——为什么 vscode 不给用了这玩意的函数做补全了啊……

让我们继续思考。

你需要什么？

vscode 的补全。

再具体一点。

我们需要让 vscode 知道新函数是什么，或者说，我们需要注解返回的类型。

那我们有什么了呢？

传入的函数的类型。

所以这个问题也很简单，我们需要根据传入函数的类型，确定传出函数的类型，显然这需要泛型。

# 这里使用了 python3.12+ 的泛型语法，旧版本需要使用 TypeVar 和 ParamSpec
from typing import Callable,Generator,ContextManager
def contextmanager[**FuncArgs,YieldType](func:Callable[FuncArgs,Generator[YieldType,None,None]])->Callable[FuncArgs,ContextManager[YieldType]]:
    class Wrapper:
        def __init__(self,*args,**kwargs):
            self.g=func(*args,**kwargs)
        def __enter__(self):
            return next(self.g)
        def __exit__(self, exc_type, exc_value, traceback):
            try:
                next(self.g)
            except StopIteration:
                pass
    return Wrapper

既然说到类型了，那么不妨试试这些代码吧：

from typing import Callable,Generator,ContextManager
def contextmanager[**FuncArgs,YieldType](func:Callable[FuncArgs,Generator[YieldType,None,None]])->Callable[FuncArgs,ContextManager[YieldType]]:
    class Wrapper:
        def __init__(self,*args,**kwargs):
            self.g=func(*args,**kwargs)
        def __enter__(self):
            return next(self.g)
        def __exit__(self, exc_type, exc_value, traceback):
            try:
                next(self.g)
            except StopIteration:
                pass
    return Wrapper

@contextmanager
def fun(x:int)->Generator[int, None, None]:
    yield x

help(fun)

发现问题了吗？我们的东西装饰完之后，函数完全不能看了。

我们需要什么呢？

能看的函数，准确的说，运行的时候保留函数签名。

我们有什么呢？

原来的函数，这里留着原来的签名。

显然，我们需要修改新函数的签名，让它跟原来的相同。

这里不卖关子了，functools 库提供了这类工具。

只要这样就好了：

from typing import Callable,Generator,ContextManager
import functools
def contextmanager[**FuncArgs,YieldType](func:Callable[FuncArgs,Generator[YieldType,None,None]])->Callable[FuncArgs,ContextManager[YieldType]]:
    class Wrapper:
        def __init__(self,*args,**kwargs):
            self.g=func(*args,**kwargs)
        def __enter__(self):
            return next(self.g)
        def __exit__(self, exc_type, exc_value, traceback):
            try:
                next(self.g)
            except StopIteration:
                pass
    @functools.wraps(func)
    def wrapper(*args,**kwargs):
        return Wrapper(*args,**kwargs)
    return wrapper

@contextmanager
def fun(x:int)->Generator[int, None, None]:
    yield x

help(fun)

:::warning[注意]{open}

@functools.wraps(func)
def wrapper(*args,**kwargs):
    return Wrapper(*args,**kwargs)

这里额外包装了 Wrapper，就是因为函数和可调用对象（一个有 __init__ 方法的类）的细节区别。

如果直接对类使用的话会有类似这样的报错：

Traceback (most recent call last):
  File "<python-input-4>", line 18, in <module>
    @contextmanager
     ^^^^^^^^^^^^^^
  File "<python-input-4>", line 5, in contextmanager
    @functools.wraps(func)
     ~~~~~~~~~~~~~~~^^^^^^
  File "/usr/lib64/python3.14/functools.py", line 58, in update_wrapper
    getattr(wrapper, attr).update(getattr(wrapped, attr, {}))
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'mappingproxy' object has no attribute 'update'

因为 wraps 会尝试同步对象的 __dict__ 属性，而 python 中，类的 __dict__ 是 mappingproxy 的，mappingproxy 不允许通过 .update() 修改。

functools.wraps() 的部分源码在文末。 :::

现在，正常的 help 是不是回来了？

3

这看起来的确不错，也似乎能直接拿来用了。让我们尝试用它写点东西吧。

不如让我们写另一个 contextlib 里面的东西——suppress？

让我们尝试按照之前那样分析吧？

我们需要接收一个错误类型，并且只有在得到这个类型的错误的时候不向上抛出。所以，我们需要知道怎么拿到错误，以及怎么拦截它们。

是啊，怎么拦截它们呢？你肯定发现问题了，我们根本没设计这个接口……

看来我们有更大的问题了，我们需要设计个接口来处理 with 的异常，并且把这点东西给生成器函数。

那么，我们需要知道怎么从 with 手里拿到异常，怎么告诉 with 异常已经被我们截胡了，另外，我们还需要知道怎么把异常给生成器。

那么，这就是我们需要的资料：

上下文管理器（提示 2）

对于上下文管理器，刚刚省略了一点，细心的同学肯定发现了——为什么 __exit__ 函数有那么多参数啊？

没错，这些就是用来处理异常的。

__exit__(self, exc_type, exc_value, traceback) 方法在离开 with 块的时候调用。如果 with 中发生异常，会作为参数传入 __exit__，如果函数返回 True 则代表异常被处理，with 不会向上抛出它；否则返回 False 或 None 则 with 会重新抛出异常。

生成器（提示 2）

如果你写过其它语言的异常处理的话，应该会看到生成器的函数中有一个明晃晃的 throw，没错，这就是用来放入异常的。

>>> help(g.throw)

Help on method descriptor throw:

throw(...) unbound builtins.generator method
    throw(value)
    throw(type[,value[,tb]])

    Raise exception in generator, return next yielded value or raise
    StopIteration.
    the (type, val, tb) signature is deprecated,
    and may be removed in a future version of Python.

throw() 传入异常，异常会在函数内当前 yield 抛出，然后 throw() 返回下一个 yield 的值。

这玩意怎么这么像异步啊。

这个方法也会在函数结束时抛出 StopIteration 异常。

改一改

那么思路应该很明确了，如果 __exit__ 传入了异常，丢给生成器，生成器处理了就告诉 with 不要抛出。

from typing import Callable,Generator,ContextManager
import functools
def contextmanager[**FuncArgs,YieldType](func:Callable[FuncArgs,Generator[YieldType,None,None]])->Callable[FuncArgs,ContextManager[YieldType]]:
    class Wrapper:
        def __init__(self,*args,**kwargs):
            self.g=func(*args,**kwargs)
        def __enter__(self):
            return next(self.g)
        def __exit__(self, exc_type, exc_value, traceback):
            if exc_type is None:
                try:
                    next(self.g)
                except StopIteration as _:
                    return False
            else:
                if exc_value is None:
                    exc_value=exc_type()
                try:
                    self.g.throw(exc_value)
                except StopIteration as e:
                    return True
                except BaseException as e:
                    return False
    @functools.wraps(func)
    def wrapper(*args,**kwargs):
        return Wrapper(*args,**kwargs)
    return wrapper

那么根据我们自己的接口，写一个 suppress 应该也不难。

@contextmanager
def suppress(e:type[BaseException]):
    try:
        yield
    except e:
        pass

4

如果你真的用过 @contextmanager 的话，应该知道，它还有另一个小功能。

使用 @contextmanager 的函数也可以作为装饰器使用：

@managed_resource()
def fun():
    ...
fun()

等价于：

def fun():
    ...
with managed_resource():
    fun()

分析一下，也就是说，我们需要让新函数的返回值可以做装饰器，也就是说，让它的返回值可以调用。

如果你能看到这里，代码应该不难想到。

from typing import Callable,Generator,ContextManager,Protocol
import functools
class WrapperProtocol[YieldType](ContextManager[YieldType],Protocol):
    def __call__[**A,T](self,func:Callable[A,T])->Callable[A,T]:
        ...
def contextmanager[**FuncArgs,YieldType](func:Callable[FuncArgs,Generator[YieldType,None,None]])->Callable[FuncArgs,WrapperProtocol[YieldType]]:
    class Wrapper:
        def __init__(self,*args,**kwargs):
            self.g=func(*args,**kwargs)
        def __enter__(self):
            return next(self.g)
        def __exit__(self, exc_type, exc_value, traceback):
            if exc_type is None:
                try:
                    next(self.g)
                except StopIteration as _:
                    return False
            else:
                if exc_value is None:
                    exc_value=exc_type()
                try:
                    self.g.throw(exc_value)
                except StopIteration as e:
                    return True
                except BaseException as e:
                    return False
        def __call__[**A,T](self,func:Callable[A,T])->Callable[A,T]:
            def wrapper(*args,**kwargs):
                with self:
                    return func(*args,**kwargs)
            return functools.update_wrapper(wrapper,func)
    @functools.wraps(func)
    def wrapper(*args,**kwargs):
        return Wrapper(*args,**kwargs)
    return wrapper

让我们试用一下吧：

@contextmanager
def say_hello():
    print("hello")
    try:
        yield
    finally:
        print("bye")

@say_hello()
def say_something(s:str):
    print(s)

>>> say_something("十年 OI 一场空，不开 long long 见祖宗")
hello
十年 OI 一场空，不开 long long 见祖宗
bye
>>> 

它真的跑起来了——多么值得兴奋地一件事啊，它居然能跑……让我们多试几次好好享受一下今天的成果吧……

>>> say_something("十年 OI 一场空，不开 long long 见祖宗")
Traceback (most recent call last):
  File "<python-input-7>", line 1, in <module>
    say_something("十年 OI 一场空，不开 long long 见祖宗")
    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<python-input-1>", line 27, in wrapper
    with self:
         ^^^^
  File "<python-input-1>", line 11, in __enter__
    return next(self.g)
StopIteration

哎？不对，怎么真的见祖宗了?

StopIteration，这应该是迭代器结束之后调用 next 或者 throw 的结果……看起来我们在迭代器结束之后把它当成没结束调用了一次。

确实如此，我们只在 __init__ 的时候拿到了迭代器，但是却尝试把它用了两次，当然会炸，解决方案也很简单，记住参数，再搞一个迭代器就好了。

from typing import Callable,Generator,ContextManager,Protocol
import functools
class WrapperProtocol[YieldType](ContextManager[YieldType],Protocol):
    def __call__[**A,T](self,func:Callable[A,T])->Callable[A,T]:
        ...
def contextmanager[**FuncArgs,YieldType](func:Callable[FuncArgs,Generator[YieldType,None,None]])->Callable[FuncArgs,WrapperProtocol[YieldType]]:
    class Wrapper:
        def __init__(self,*args,**kwargs):
            self.g=func(*args,**kwargs)
            self.args=args
            self.kwargs=kwargs
        def __enter__(self):
            return next(self.g)
        def __exit__(self, exc_type, exc_value, traceback):
            if exc_type is None:
                try:
                    next(self.g)
                except StopIteration as _:
                    return False
            else:
                if exc_value is None:
                    exc_value=exc_type()
                try:
                    self.g.throw(exc_value)
                except StopIteration as e:
                    return True
                except BaseException as e:
                    return False
        def __call__[**A,T](self,func:Callable[A,T])->Callable[A,T]:
            def wrapper(*args,**kwargs):
                with self.__class__(*self.args,**self.kwargs):
                    return func(*args,**kwargs)
            return functools.update_wrapper(wrapper,func)
    @functools.wraps(func)
    def wrapper(*args,**kwargs):
        return Wrapper(*args,**kwargs)
    return wrapper

:::info[不要点开] Gemini 说这里 __init__ 会多获取一遍生成器，但是我不打算修复，因为正版也没管。 :::

>>> @contextmanager
... def say_hello():
...     print("hello")
...     try:
...         yield
...     finally:
...         print("bye")
... 
... @say_hello()
... def say_something(s:str):
...     print(s)
...     
>>> say_something("十年 OI 一场空，不开 long long 见祖宗")
hello
十年 OI 一场空，不开 long long 见祖宗
bye
>>> say_something("十年 OI 一场空，不开 long long 见祖宗")
hello
十年 OI 一场空，不开 long long 见祖宗
bye
>>> 

5

还有一些情况我们仍然没考虑到：

如果迭代器没有正常结束呢？

如果迭代器没 yield 就结束了呢？

如果迭代器自己抛出了一个错误呢？

如果迭代器把我们丢进去的错误原样丢出来了呢？

如果 with 丢进来一个 StopIteration 呢？

如果 __enter__ 了之后，又尝试把它用作迭代器了怎么办？

错误信息是否完整易读？

…………

这些可以尝试自己思考一下。

当然，也强烈推荐阅读 contextlib 的源码（文末截取了和 contextmanager 有关的部分），尤其应该看看它的注释。

这点问题很多来自那点注释。

来自标准库的警示后人（）

完整代码还是不放了，写出来也跟标准库差不多……

一点笔记

contextlib.contextmanager

参考官方文档。

像这样写一个生成器函数，并且用 @contextmanager 装饰：

from contextlib import contextmanager

@contextmanager
def managed_resource(*args, **kwds):
    # Code to acquire resource, e.g.:
    resource = acquire_resource(*args, **kwds)
    try:
        yield resource
    finally:
        # Code to release resource, e.g.:
        release_resource(resource)

然后就可以把它当成上下文管理器来用：

with managed_resource(timeout=3600) as resource:
    # Resource is released at the end of this block,
    # even if code in the block raises an exception

使用 @contextmanager 的函数也可以作为装饰器使用：

@managed_resource()
def fun():
    ...
fun()

等价于：

def fun():
    ...
with managed_resource():
    fun()

如果 with 块中出现异常，异常会在 yield 位置抛出，如果异常未被处理，那么 with 块会向上抛出异常。

如果函数中抛出异常，with 块会向上抛出。

如果生成器提前结束（一个 yield 也没有）或是结束晚（不只遇到一个 yield）会抛出 RuntimeError。

装饰器

一个函数，接受另一个可调用对象作为参数，通常我们也会返回一个可调用对象。

装饰器语法会将被装饰的对象传入函数，并且将返回值赋值给原本的名称。

@decorator
def fun():
    ...

就等价于：

def fun()：
    ...

fun=decorator(fun)

生成器

生成器函数，一种调用了返回一个生成器的函数，你只要在函数里面写 yield 它就是生成器函数。

__next__ 用来往后获取一个 yeild 的值，通常情况下，你应该用 next(obj) 来调用它。

throw() 传入异常，异常会在函数内当前 yield 抛出，然后 throw() 返回下一个 yield 的值。

这两个方法都会在函数结束时抛出 StopIteration 异常。

:::info[异步的事]{open} 早年间 python 的异步就是拿生成器做的。

如果你注意前文的 dir() 结果的话，会发现有一个 send() 方法，这玩意就是用来给生成器里面送东西的。又或者说，调度器给异步函数送值的。

函数可以用类似这样的语法接收一个值:

res=yield val

send() 其它方面表现的和 __next__() 差不多。 :::

上下文管理器

常见的语法是这样的：

with context_manager as x:
    ...

基本上，一个对象有 __enter__ 和 __exit__ 就能被称为上下文管理器了。

__enter__() 会在进入 with 块的时候调用，其返回值会赋值给 as 后的名称（如果有 as 的话）。

__exit__(self, exc_type, exc_value, traceback) 方法在离开 with 块的时候调用。如果 with 中发生异常，会作为参数传入 __exit__，如果函数返回 True 则代表异常被处理，否则返回 False 或 None 则 with 会重新抛出异常。

部分源码

节选自 python3.14 的 functools.py，截取了和 wraps 有关的部分。

WRAPPER_ASSIGNMENTS = ('__module__', '__name__', '__qualname__', '__doc__',
                       '__annotate__', '__type_params__')
WRAPPER_UPDATES = ('__dict__',)
def update_wrapper(wrapper,
                   wrapped,
                   assigned = WRAPPER_ASSIGNMENTS,
                   updated = WRAPPER_UPDATES):
    """Update a wrapper function to look like the wrapped function

       wrapper is the function to be updated
       wrapped is the original function
       assigned is a tuple naming the attributes assigned directly
       from the wrapped function to the wrapper function (defaults to
       functools.WRAPPER_ASSIGNMENTS)
       updated is a tuple naming the attributes of the wrapper that
       are updated with the corresponding attribute from the wrapped
       function (defaults to functools.WRAPPER_UPDATES)
    """
    for attr in assigned:
        try:
            value = getattr(wrapped, attr)
        except AttributeError:
            pass
        else:
            setattr(wrapper, attr, value)
    for attr in updated:
        getattr(wrapper, attr).update(getattr(wrapped, attr, {}))
    # Issue #17482: set __wrapped__ last so we don't inadvertently copy it
    # from the wrapped function when updating __dict__
    wrapper.__wrapped__ = wrapped
    # Return the wrapper so this can be used as a decorator via partial()
    return wrapper

def wraps(wrapped,
          assigned = WRAPPER_ASSIGNMENTS,
          updated = WRAPPER_UPDATES):
    """Decorator factory to apply update_wrapper() to a wrapper function

       Returns a decorator that invokes update_wrapper() with the decorated
       function as the wrapper argument and the arguments to wraps() as the
       remaining arguments. Default arguments are as for update_wrapper().
       This is a convenience function to simplify applying partial() to
       update_wrapper().
    """
    return partial(update_wrapper, wrapped=wrapped,
                   assigned=assigned, updated=updated)

节选自 python3.14 的 contextlib.py，截取了和 contextmanager 有关的部分。

import abc
import os
import sys
import _collections_abc
from collections import deque
from functools import wraps
from types import MethodType, GenericAlias

class AbstractContextManager(abc.ABC):

    """An abstract base class for context managers."""

    __class_getitem__ = classmethod(GenericAlias)

    __slots__ = ()

    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

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 _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)
        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 http://bugs.python.org/issue19404 for more details.

    def _recreate_cm(self):
        # _GCMB 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)

class _GeneratorContextManager(
    _GeneratorContextManagerBase,
    AbstractContextManager,
    ContextDecorator,
):
    """Helper for @contextmanager decorator."""

    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, typ, value, traceback):
        if typ is None:
            try:
                next(self.gen)
            except StopIteration:
                return False
            else:
                try:
                    raise RuntimeError("generator didn't stop")
                finally:
                    self.gen.close()
        else:
            if value is None:
                # Need to force instantiation so we can reliably
                # tell if we get the same exception back
                value = typ()
            try:
                self.gen.throw(value)
            except StopIteration as exc:
                # 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:
                # Don't re-raise the passed in exception. (issue27122)
                if exc is value:
                    exc.__traceback__ = traceback
                    return False
                # Avoid suppressing if a StopIteration exception
                # was passed to throw() and later wrapped into a RuntimeError
                # (see PEP 479 for sync generators; async generators also
                # have this behavior). But do this only if the exception wrapped
                # by the RuntimeError is actually Stop(Async)Iteration (see
                # issue29692).
                if (
                    isinstance(value, StopIteration)
                    and exc.__cause__ is value
                ):
                    value.__traceback__ = traceback
                    return False
                raise
            except BaseException as exc:
                # 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.
                if exc is not value:
                    raise
                exc.__traceback__ = traceback
                return False
            try:
                raise RuntimeError("generator didn't stop after throw()")
            finally:
                self.gen.close()

def contextmanager(func):
    """@contextmanager decorator.

    Typical usage:

        @contextmanager
        def some_generator(<arguments>):
            <setup>
            try:
                yield <value>
            finally:
                <cleanup>

    This makes this:

        with some_generator(<arguments>) as <variable>:
            <body>

    equivalent to this:

        <setup>
        try:
            <variable> = <value>
            <body>
        finally:
            <cleanup>
    """
    @wraps(func)
    def helper(*args, **kwds):
        return _GeneratorContextManager(func, args, kwds)
    return helper

:::info[AI 使用说明]{open} 本文在创作过程中使用了生成式 AI

使用 Gemini 用于查找部分资料、检查错误、以及一些润色建议。 :::