标签: contextmanager

将上下文管理器定义为函数，可以很容易地以编程方式从其中输入一个单独的（或递归的）上下文管理器，如下所示：

@contextmanager
def enter(times):
    if times:
        with enter(times - 1) as tup:
            print 'entering {}'.format(times)
            yield tup + (times,)
            print 'exiting {}'.format(times)
    else:
        yield ()

运行这个：

In [11]: with enter(4) as x:
....:     print x
....:
entering 1
entering 2
entering 3
(1, 2, 3)
exiting 3
exiting 2
exiting 1

所有出入境记账都为您完成，多好啊！但是如果你有一个类，而不是一个函数怎么办？

class Enter(object):
    def __init__(self, times):
        self.times = times

    def __enter__(self):
        print 'entering {}'.format(self.times)
        if self.times:
            with Enter(self.times - 1) as tup:  # WRONG
                return tup + (self.times,)
        return …

我想要一个 pyplot 图形的上下文管理器，基本上像这样：

from contextlib import contextmanager
import matplotlib.pyplot as plt

@contextmanager
def subplots():
  (fig, ax) = plt.subplots()
  try:
    yield (fig, ax)
  finally:
    plt.close(fig)

是否可以对返回的元组实现类型提示？天真的

import typing
def subplots() -> typing.Tuple[plt.Figure, plt.Axes]

不起作用。

我正在学习上下文管理器，并试图自己构建一个。以下是一个虚拟上下文管理器，它以读取模式打开一个文件（我知道我可以这样做with open(...): ...。这只是我构建的一个示例，旨在帮助我了解如何创建自己的上下文管理器）：

@contextmanager
def open_read(path: str):
    f = open(path, 'r')
    print('open')
    yield f
    f.close()
    print('closed')


def foo():
    try:
        with open_read('main.py') as f:
            print(f.readline())
            raise Exception('oopsie')
    except Exception:
        pass
    print(f.readline())


foo()

我希望这段代码能够打印：

open
<line 1 of a.txt>
closed
ValueError: I/O operation on closed file.

但它打印的是：

open
<line 1 of a.txt>
<line 2 of a.txt>

它没有关闭文件！

这似乎与 python 的文档相矛盾，该文档声明无论语句成功退出还是出现异常__exit__都会调用该函数：with

目的。退出（自身、exc_type、exc_value、回溯）

退出与该对象相关的运行时上下文。参数描述导致上下文退出的异常。如果上下文退出时没有异常，则所有三个参数都将为 None。

有趣的是，当我重新实现上下文管理器（如下所示）时，它按预期工作：

open
<line 1 of a.txt>
closed
ValueError: I/O operation on closed …

下面粘贴的代码执行以下操作:

• 创建一个导入钩子
• 创建一个上下文管理器,用于设置meta_path和清除退出.
• 转储在imports.log中输入的程序完成的所有导入

现在我想知道在这种情况下使用上下文管理器是否是一个好主意,因为实际上我没有标准try/finally流程,只是设置和清理.

另一件事 - 用这条线:

with CollectorContext(cl, sys.argv, 'imports.log') as cc:

确实cc成了None？它不应该是一个CollectorContext对象吗？

from __future__ import with_statement
import os
import sys

class CollectImports(object):
    """
    Import hook, adds each import request to the loaded set and dumps
    them to file
    """

    def __init__(self):
        self.loaded = set()

    def __str__(self):
        return str(self.loaded)

    def dump_to_file(self, fname):
        """Dump the loaded set to file
        """
        dumped_str = '\n'.join(x for x in self.loaded)
        open(fname, 'w').write(dumped_str) …

我有一个对象,我希望能够使用with关键字.我对实施上下文管理器的实际情况感到满意,但我正面对最佳实践问题.

该对象是文件的包装器.我计划用一个字符串(文件的路径)或者像文件一样初始化我的对象,这可以直接处理(文件里面有文件的可能性 - 所以我预见到一个明确的用例这与BytesIO等...)

所以__init__看起来像这样:

def __init__(self, file_to_process):
    if isinstance(file_to_process, str):
        self._underlying_stream = open(file_to_process, "rb") # it's the path to a file
    elif isinstance(file_to_process, io.IOBase):
        self._underlying_stream = file_to_process # its the file itself
    else:
         raise TypeError()

所以我的问题是,_underlying_stream在我的__exit__()功能中关闭它是最好的做法/接受/明智吗？当它成为一条小路时,它完全有道理,但如果它是一条小溪传入,它最让我感到不礼貌,最坏的情况是关闭self._underlying_stream- 我是否正确思考,如果是这样,是否有一个整洁的方式这个？

(注意:我考虑用a来包装流io.BufferedReader,但事实证明,关闭也将关闭底层流...)

我想通过类上下文管理器强制对象实例化。因此，使其无法直接实例化。

我实现了此解决方案，但从技术上讲，用户仍然可以实例化对象。

class HessioFile:
    """
    Represents a pyhessio file instance
    """
    def __init__(self, filename=None, from_context_manager=False):
        if not from_context_manager:
            raise HessioError('HessioFile can be only use with context manager')

和上下文管理器：

@contextmanager
def open(filename):
    """
    ...
    """
    hessfile = HessioFile(filename, from_context_manager=True)

有更好的解决方案吗？

假设我有一些我正在使用的上下文管理器(来自第三方库):

with freeze_time(test_dt):
    lines_of_code_1
    lines_of_code_2
    lines_of_code_3

但是,假设如果test_dt没有值,则上下文管理器不应该运行,但是所有剩余的代码都应该运行,如下所示:

if test_dt:
    with freeze_time(test_dt):
        lines_of_code_1
        lines_of_code_2
        lines_of_code_3
else:
    lines_of_code_1
    lines_of_code_2
    lines_of_code_3

假设lines_of_code这里有2-3行完全相同的代码,是否有更清晰的写法呢？我知道我可以这样写:

def do_thing():
    lines_of_code_1
    lines_of_code_2
    lines_of_code_3

if test_dt:
    with freeze_time(test_dt):
        do_thing()
else:
    do_thing()

但我对这种格式并不感到疯狂.另外,我不想在我的代码中乱丢这个模式.

最后一种可能性,但我不确定它是否会起作用:如果给出的test_dt为空,则继承上下文管理器并跳过__enter__和__exit__函数,如下所示:

class optional_freeze_time(object):
    def __init__(self, test_dt=None):
        if test_dt:
            self.ctx_manager = freeze_time(test_dt)
        else:
            self.ctx_manager = None
    def __enter__(self, *args, **kwargs):
        if self.ctx_manager:
            self.ctx_manager.__enter__(*args, **kwargs)
    def __exit__(self, *args, **kwargs):
        if self.ctx_manager:
            self.ctx_manager.__exit__(*args, **kwargs)

我用一个空白的上下文管理器类测试了它,它似乎行为正确.但是,如果我这样做,我担心真正的上下文管理器是否会正常运行(我对它的工作方式不太熟悉).

AttributeError: __enter__当我尝试使用像本指南中的 sqlalchemy会话时,我得到这个.

我的代码:

Session = scoped_session(sessionmaker(autoflush=True, autocommit=False, bind=engine))

@contextmanager
def session_scope():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()

class SomeClass:

    def __init__(self):
        self.session_scope = session_scope

    def something_with_session(self):

        with self.session_scope as session: <-- error

我究竟做错了什么？我正在使用python 3.6

下面是我的my_create方法的示例,以及使用该方法的示例.

@contextmanager
def my_create(**attributes):
    obj = MyObject(**attributes)
    yield obj
    obj.save()

with my_create(a=10) as new_obj:
     new_obj.b = 7

new_obj.a  # => 10
new_obj.b  # => 7
new_obj.is_saved()  # => True

对Ruby/Rails的用户来说,这可能看起来很熟悉.它类似于ActiveRecord::create方法,with块内的代码就像一个块一样.

然而:

with my_create(a=10) as new_obj:
    pass

new_obj.a  # => 10
new_obj.is_saved()  # => True

在上面的例子中,我将一个空的"块"传递给了我的my_create函数.事情按预期工作(my_obj已初始化并保存),但格式看起来有点不稳定,并且with块似乎没有必要.

我宁愿能够my_create直接打电话,而不必设置passing with块.不幸的是,我当前的实现不可能my_create.

my_obj = create(a=10)
my_obj  # => <contextlib.GeneratorContextManager at 0x107c21050>

我不得不呼吁双方__enter__并__exit__在 …

我不明白如何以及何时关闭未完成的生成器中的上下文管理器.考虑以下上下文管理器和功能:

from contextlib import contextmanager

@contextmanager
def ctx():
    print('enter ctx')
    yield
    print('exit ctx')

def gen_nums(n):
    with ctx():
        yield from range(n)

我的第一个直觉是,如果我打电话gen_nums但不完全消耗发电机,ctx永远不会关闭,这是相当令人担忧的.例如:

for i, j in zip(range(5), gen_nums(10)):
    print(f'{i}, {j}')

这里exit ctx是未在结束打印.正如我所看到的,这意味着如果我在生成器中有一个文件上下文,它将保持打开状态; 然而,我意识到对文件做同样的事情实际上会正确地关闭文件.经过一些测试,我发现如果我做了:

from contextlib import contextmanager

@contextmanager
def ctx():
    print('enter ctx')
    try:
        yield
    finally:
        print('exit ctx')

现在exit ctx最后印了.所以我想在某些时候会触发一些异常,但我不知道哪个,哪里或何时(我试图打印异常,except BaseException as e但它不起作用).它似乎发生在删除生成器时,因为如果我这样做:

g = gen_nums(10)
for i, j in zip(range(5), g):
    print(f'{i}, {j}')
del g

然后exit ctx才发生 …