标签: contextmanager

我正在尝试编写一个使用其他上下文管理器的上下文管理器,因此客户端不需要知道整个配方,只需要知道我正在呈现的界面.我无法使用@contextmanager- yield如果您被异常中断,则调用后的代码不会被执行,因此我需要使用基于类的管理器.

这是一个小例子脚本:

from contextlib import contextmanager
import pprint

d = {}

@contextmanager
def simple(arg, val):
    print "enter", arg
    d[arg] = val
    yield
    print "exit", arg
    del d[arg]

class compl(object):
    def __init__(self, arg, val):
        self.arg=arg
        self.val=val

    def __enter__(self):
        with simple("one",1):
            with simple("two",2):
                print "enter complex", self.arg
                d[self.arg] = self.val

    def __exit__(self,*args):
        print "exit complex", self.arg
        del d[self.arg]

print "before"
print d
print ""

with compl("three",3):
    print d
    print ""

print "after"
print d
print ""

这输出:

before …

使用装饰器为函数的返回添加类型提示的正确方法是什么@asynccontextmanager？这是我所做的两次尝试，但都失败了。

from contextlib import asynccontextmanager
from typing import AsyncContextManager


async def caller():
    async with return_str() as i:
        print(i)

    async with return_AsyncContextManager() as j:
        print(j)


@asynccontextmanager
async def return_str() -> str:
    yield "hello"


@asynccontextmanager
async def return_AsyncContextManager() -> AsyncContextManager[str]:
    yield "world"

对于 vscode 中的 Pylancei和j Pylance显示类型Any。我考虑过的其他想法：

• 我想也许我可以将类型信息传递给装饰器本身（例如@asynccontextmanager(cls=str)，但我找不到任何示例，或者我可以传递的参数的任何描述。
• async with return_str() as i: # type: str也不行。即使确实如此，我也宁愿暗示函数定义，而不是在每次调用时都暗示。类型注释不是很 DRY。
• 我尝试创建一个具有和函数AsyncContextManager的类，但没有成功。如果它有效的话我会回到那个，但我更喜欢让装饰器工作，因为它更简洁。__aenter()____aexit()__

这是我将鼠标悬停在该return_AsyncContextManager()函数上的屏幕截图，并显示 Pylance 弹出窗口显示它返回AsyncContextManager[_T]

这是我的代码的简化版本：

main是一个在第二次迭代后停止的协程。
get_numbers是一个异步生成器，它生成数字，但位于异步上下文管理器中。

import asyncio


class MyContextManager:
    async def __aenter__(self):
        print("Enter to the Context Manager...")
        return self

    async def __aexit__(self, exc_type, exc_value, exc_tb):
        print(exc_type)
        print("Exit from the Context Manager...")
        await asyncio.sleep(1)
        print("This line is not executed")  # <-------------------
        await asyncio.sleep(1)


async def get_numbers():
    async with MyContextManager():
        for i in range(30):
            yield i


async def main():
    async for i in get_numbers():
        print(i)
        if i == 1:
            break


asyncio.run(main())

输出是：

Enter to the Context Manager...
0
1
<class 'asyncio.exceptions.CancelledError'>
Exit …

以下是Richard Jones博客的一些代码:

with gui.vertical:
    text = gui.label('hello!')
    items = gui.selection(['one', 'two', 'three'])
    with gui.button('click me!'):
        def on_click():
            text.value = items.value
            text.foreground = red

我的问题是:他是怎么做到的？上下文管理器如何访问with块内的范围？这是一个试图解决这个问题的基本模板:

from __future__ import with_statement

class button(object):
  def __enter__(self):
    #do some setup
    pass

  def __exit__(self, exc_type, exc_value, traceback):
    #XXX: how can we find the testing() function?
    pass

with button():
  def testing():
    pass

据我了解，Python 中使用上下文管理器来定义对象的初始化和完成代码段 (__enter__和)。__exit__

然而，在PyMC3 教程中，他们展示了以下上下文管理器示例：

basic_model = pm.Model()

with basic_model:

    # Priors for unknown model parameters
    alpha = pm.Normal('alpha', mu=0, sd=10)
    beta = pm.Normal('beta', mu=0, sd=10, shape=2)
    sigma = pm.HalfNormal('sigma', sd=1)

    # Expected value of outcome
    mu = alpha + beta[0]*X1 + beta[1]*X2

    # Likelihood (sampling distribution) of observations
    Y_obs = pm.Normal('Y_obs', mu=mu, sd=sigma, observed=Y)

并提到这的目的是将变量alpha、beta、sigma和mu与Y_obs模型 basic_model 相关联。

我想了解这样的机制是如何运作的。在我发现的上下文管理器的解释 中，我没有看到任何建议上下文块中定义的变量或对象如何以某种方式“关联”到上下文管理器。库（PyMC3）似乎可以以某种方式访问​​“当前”上下文管理器，因此它可以在幕后将每个新创建的语句与其关联。但是库如何访问上下文管理器呢？

from contextlib import contextmanager

@contextmanager
def context():
    print "entering"
    yield
    print "exiting"

def test():
    with context():
        for x in range(10):
            yield x

for x in test():
    if x == 5:
        break  # or raise

输出:

entering

__exit__当for-loop被中断时,有没有办法让python自动调用context()的方法？或者其他一些实现相同目标的方式？我对生成器和上下文管理器的了解让我怀疑它是不可能的,但是这使得上下文管理器在内部生成器中相当无用,不是吗？在我看来,块yield内的语句with应该引发一个红旗,上下文管理器__exit__可能无法运行.

众所周知,python __del__方法不应该用于清理重要的东西,因为不能保证调用此方法.另一种方法是使用上下文管理器,如几个线程中所述.

但我不太明白如何重写一个类来使用上下文管理器.详细说明,我有一个简单的(非工作)示例,其中包装器类打开并关闭设备,并且在任何情况下,类的实例都超出其范围(异常等)时将关闭设备.

第一个文件mydevice.py是打开和关闭设备的标准包装类:

class MyWrapper(object):
    def __init__(self, device):
        self.device = device

    def open(self):
        self.device.open()

    def close(self):
        self.device.close()

    def __del__(self):
        self.close()

这个类由另一个类使用myclass.py:

import mydevice


class MyClass(object):

    def __init__(self, device):

        # calls open in mydevice
        self.mydevice = mydevice.MyWrapper(device)
        self.mydevice.open()

    def processing(self, value):
        if not value:
            self.mydevice.close()
        else:
            something_else()

我的问题:当我mydevice.py用with __enter__和__exit__方法实现上下文管理器时,如何处理这个类myclass.py？我需要做点什么

def __init__(self, device):
    with mydevice.MyWrapper(device):
        ???

但是如何处理呢？也许我忽略了一些重要的事情？或者我可以仅在函数内使用上下文管理器而不是作为类范围内的变量？

我想要这样的东西:

from contextlib import contextmanager

@contextmanager
def loop(seq):
    for i in seq:
        try:
            do_setup(i)
            yield # with body executes here
            do_cleanup(i)
        except CustomError as e:
            print(e)

with loop([1,2,3]):
    do_something_else()
    do_whatever()

但是contextmanager不起作用,因为它期望生成器只生成一次.

我想要这个的原因是因为我基本上想要自己定制循环.我有一个改进的IPython,用于控制测试设备.它显然是一个完整的Python REPL,但大多数时候用户只是调用预定义函数(类似于Bash提示符),并且用户不应该是程序员或熟悉Python.需要有一种方法可以通过每次迭代的setup/cleanup和异常处理来循环一些任意代码,并且它应该像上面的with语句一样简单.

在python中,有许多函数可用作标准函数和上下文管理器.例如,open()可以称为:

my_file=open(filename,'w')

要么

with open(filename,'w') as my_file:

两者都给你一个my_file可以用来做任何你需要的对象.一般情况下,后者是优选的,但有时也可能想要做前者.

我已经能够弄清楚如何编写上下文管理器,或者通过创建一个带有__enter__和__exit__函数的类,或者通过@contextlib.contextmanager在函数上使用装饰器yield而不是return.然而,当我这样做时,我不能再直接使用该函数 - 使用装饰器,例如,我得到一个_GeneratorContextManager对象而不是想要的结果.当然,如果我把它作为一个类,我只会得到一个生成器类的实例,我假设它基本上是相同的.

那么我如何设计一个函数(或类)作为函数,返回一个对象或一个上下文管理器,返回一个_GeneratorContextManager或类似的东西？

编辑:

例如,假设我有一个类似下面的函数(这是高度简化的):

def my_func(arg_1,arg_2):
    result=arg_1+arg_2
    return my_class(result)

所以函数需要一些参数,用它们做些什么,并使用那些东西的结果初始化一个类,然后返回它.最终结果是我有一个实例my_class,就像我有一个file对象,如果我已经调用open.如果我希望能够将此函数用作上下文管理器,我可以像这样修改它:

@contextlib.contextmanager
def my_func(arg_1,arg_2):
    result=arg_1+arg_2 # This is roughly equivalent to the __enter__ function
    yield my_class(result)
    <do some other stuff here> # This is roughly equivalent to the __exit__function

当作为上下文管理器调用时,它工作得很好,但我不再获得my_class调用为直接函数的实例.也许我只是做错了什么？

编辑2:

请注意,我确实可以完全控制my_class,包括向其添加功能的功能.从下面的接受的答案,我可以推断,我的困难来自于一个基本的误解朵朵:我在想,不管我叫(my_func …

请考虑使用Python 2.x代码段.

from __future__ import print_function


class myfile(file):
    def __exit__(self, *excinfo):
        print("__exit__ called")
        super(myfile, self).__exit__(*excinfo)


def my_generator(file_name):
    with myfile(file_name) as fh:
        for line in fh:
            yield line.strip()


gen = my_generator('file.txt')
print(next(gen))
print("Before del")
del gen
print("After del")

此脚本的输出(给定file.txt有多行)是:

Line 1 from file
Before del
__exit__ called
After del

我__exit__特别感兴趣.

什么触发了他的方法的执行？对于我们所知道的,代码从未离开过with语句(它在yield语句后"停止" 并且从未继续).__exit__当发电机的参考计数降至0时,是否保证会被调用？