标签: contextmanager

我习惯(被宠坏了？)python的SQLite接口来处理SQL数据库.python的SQLite API中的一个很好的特性是"上下文管理器",即python的with声明.我通常以下列方式执行查询:

import as sqlite

with sqlite.connect(db_filename) as conn:
    query = "INSERT OR IGNORE INTO shapes VALUES (?,?);"
    results = conn.execute(query, ("ID1","triangle"))

使用上面的代码,如果我的查询修改了数据库并且我忘记运行conn.commit(),则上下文管理器在退出with语句时自动为我运行它.它还可以很好地处理异常:如果在我提交任何内容之前发生异常,则回滚数据库.

我现在正在使用MySQLdb界面,它似乎不支持开箱即用的类似上下文管理器.我如何创建自己的？还有一个相关的问题在这里,但它并没有提供一个完整的解决方案.

我最近编写了一个返回一系列打开文件的方法; 换句话说,这样的事情:

# this is very much simplified, of course
# the actual code returns file-like objects, not necessarily files
def _iterdir(self, *path):
    dr = os.path.join(*path)
    paths = imap(lambda fn: os.path.join(dr, fn), os.listdir(dr))

    return imap(open, paths)

从语法上讲,如果我执行以下操作,我不希望必须关闭生成的对象:

for f in _iterdir('/', 'usr'):
    make_unicorns_from(f)
    # ! f.close()

结果,我决定包装_iterdir一个上下文管理器:

def iterdir(self, *path):
    it = self._iterdir(*path)

    while 1:
        with it.next() as f:
            yield f

这似乎工作正常.

我感兴趣的是这样做是否是好习惯.我是否会遇到这种模式之后的任何问题(可能会抛出异常)？

我无法弄清楚如何在类中使用上下文管理器时模拟两个文件打开.我知道如何使用mock模块为一个上下文管理的文件执行此操作:

@patch('__builtin__.open')
def test_interface_mapping(self, mock_config):
        m = MagicMock(spec=file)
        handle = m.return_value.__enter__.return_value
        handle.__iter__.return_value = ('aa', 'bb')

我的问题是当一个类在同一个调用中打开两个不同的文件时如何执行此操作.在我的例子中,该类__init__()将文件预加载到两个映射中.该类用于其他类.我想模拟这两个文件的加载以提供我的测试数据,以便可以针对我预加载的测试文件内容测试使用IfAddrConfig对象的其他类.

这是我正在努力加入两个文件的类的示例__init__(),我想要模拟这两个文件加载我的测试注入文件内容.getInterfaceMap()是经常调用的函数,因此我不希望每次调用都加载和解析文件,因此__init__()一次预加载映射的原因.

class IfAddrConfig(object):
    def __init__(self):
        # Initialize the static maps once since they require file operations
        # that we do not want to be calling every time getInterfaceMap() is used
        self.settings_map = self.loadSettings()
        self.config_map = self.loadConfig()

    def loadConfig(self):
        config_map = defaultdict(dict)
        with open(os.path.join('some_path.cfg'), 'r') as stream:
            for line in stream:
                # Parse line and build up config_map entries …

好像我不明白 - 蟒蛇with声明.

考虑这个课程:

class test(object):
    def __enter__(self): pass
    def __exit__(self, *ignored): pass

现在,当使用它时with,就像在

with test() as michael:
    print repr(michael)

我希望有一些输出像<test instance at memore blah>.但是我没有.

这里有什么问题吗？任何建议都会有帮助.

(我使用的是Python 2.6.6.)

编辑:

感谢 ephement指向我的文档.该__enter__方法应阅读

    def __enter__(self): return self

我很好奇是否认为安全或良好的做法依赖于python的...作为声明.例如,打开文件时:

with open("myfile","w") as myFile:
    #do something

所以在这个例子中我忽略了显式调用myFile.close()但是我可以假设当python with...as通过调用objects __exit__()方法退出语句时调用它.依赖于此是好的做法/安全还是总是明确地打电话会更好file.close()

import contextlib
import time

@contextlib.contextmanager
def time_print(task_name):
    t = time.time()
    try:
        yield
    finally:
        print task_name, "took", time.time() - t, "seconds."


def doproc():
    x=1+1


with time_print("processes"):
    [doproc() for _ in range(500)]

# processes took 15.236166954 seconds.

什么时候doproc在使用这个装饰器时被执行？

我正在测试代码,其中可以引发两个异常之一:MachineError或NotImplementedError.我想用它pytest.raises来确保在运行我的测试代码时至少引发其中一个,但它似乎只接受一个异常类型作为参数.

这是签名pytest.raises:

raises(expected_exception, *args, **kwargs)

我尝试or在上下文管理器中使用关键字:

with pytest.raises(MachineError) or pytest.raises(NotImplementedError):
    verb = Verb("donner<IND><FUT><REL><SG><1>")
    verb.conjugate()

但我认为这只检查第一个pytest.raises是否None,并将第二个设置为上下文管理器(如果是).

将多个异常作为位置参数传递不起作用,因为pytest.raises它的第二个参数是可调用的.每个后续位置参数都作为参数传递给该可调用对象.

从文档:

>>> raises(ZeroDivisionError, lambda: 1/0)
<ExceptionInfo ...>

>>> def f(x): return 1/x
...
>>> raises(ZeroDivisionError, f, 0)
<ExceptionInfo ...>
>>> raises(ZeroDivisionError, f, x=0)
<ExceptionInfo ...>

将异常作为列表传递也不起作用:

Traceback (most recent call last):
  File "<pyshell#4>", line 1, in <module>
    with pytest.raises([MachineError, NotImplementedError]):
  File "/usr/local/lib/python3.4/dist-packages/_pytest/python.py", line 1290, in raises
    raise TypeError(msg …

我正在尝试编写一个使用其他上下文管理器的上下文管理器,因此客户端不需要知道整个配方,只需要知道我正在呈现的界面.我无法使用@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 …