检测上下文管理器嵌套

Question

最近想知道有没有办法检测上下文管理器是否嵌套。

我已经创建了 Timer 和 TimerGroup classes:

class Timer:
    def __init__(self, name="Timer"):
        self.name = name
        self.start_time = clock()

    @staticmethod
    def seconds_to_str(t):
        return str(timedelta(seconds=t))

    def end(self):
        return clock() - self.start_time

    def print(self, t):
        print(("{0:<" + str(line_width - 18) + "} >> {1}").format(self.name, self.seconds_to_str(t)))

    def __enter__(self):
        return self

    def __exit__(self, exc_type, value, traceback):
        self.print(self.end())


class TimerGroup(Timer):
    def __enter__(self):
        print(('= ' + self.name + ' ').ljust(line_width, '='))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        total_time = self.seconds_to_str(self.end())
        print(" Total: {0}".format(total_time).rjust(line_width, '='))
        print()

此代码以可读格式打印计时：

with TimerGroup("Collecting child documents for %s context" % context_name):
    with Timer("Collecting context features"):
        # some code...
    with Timer("Collecting child documents"):
        # some code...


= Collecting child documents for Global context ============
Collecting context features                >> 0:00:00.001063
Collecting child documents                 >> 0:00:10.611130
====================================== Total: 0:00:10.612292

但是，当我嵌套 TimerGroups 时，它搞砸了：

with TimerGroup("Choosing the best classifier for %s context" % context_name):
    with Timer("Splitting datasets"):
        # some code...
    for cname, cparams in classifiers.items():
        with TimerGroup("%s classifier" % cname):
            with Timer("Training"):
                # some code...
            with Timer("Calculating accuracy on testing set"):
                # some code


= Choosing the best classifier for Global context ==========
Splitting datasets                         >> 0:00:00.002054
= Naive Bayes classifier ===================================
Training                                   >> 0:00:34.184903
Calculating accuracy on testing set        >> 0:05:08.481904
====================================== Total: 0:05:42.666949

====================================== Total: 0:05:42.669078

我所需要做的就是以某种方式缩进嵌套的 Timers 和 TimerGroups。我应该将任何参数传递给它们的构造函数吗？或者我可以从 class 内部检测到吗？

Answer 1

没有检测嵌套上下文管理器的特殊工具，没有。你必须自己处理这件事。您可以在自己的上下文管理器中执行此操作：

import threading


class TimerGroup(Timer):
    _active_group = threading.local()

    def __enter__(self):
        if getattr(TimerGroup._active_group, 'current', False):
            raise RuntimeError("Can't nest TimerGroup context managers")
        TimerGroup._active_group.current = self
        print(('= ' + self.name + ' ').ljust(line_width, '='))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        TimerGroup._active_group.current = None
        total_time = self.seconds_to_str(self.end())
        print(" Total: {0}".format(total_time).rjust(line_width, '='))
        print()

然后您可以在其他地方使用 TimerGroup._active_group 属性来获取当前活动的组。我使用 thread-local object 来确保它可以跨多个执行线程使用。

或者，您可以将其设为堆栈计数器并在嵌套的 __enter__ 调用中递增和递减，或者将其设为堆栈 list 并将 self 压入那个堆栈，当你 __exit__:

时再次弹出它

import threading


class TimerGroup(Timer):
    _active_group = threading.local()

    def __enter__(self):
        if not hasattr(TimerGroup._active_group, 'current'):
            TimerGroup._active_group.current = []
        stack = TimerGroup._active_group.current
        if stack:
            # nested context manager.
            # do something with stack[-1] or stack[0]
        TimerGroup._active_group.current.append(self)

        print(('= ' + self.name + ' ').ljust(line_width, '='))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        last = TimerGroup._active_group.current.pop()
        assert last == self, "Context managers being exited out of order"
        total_time = self.seconds_to_str(self.end())
        print(" Total: {0}".format(total_time).rjust(line_width, '='))
        print()

Answer 2

import this:

Explicit is better than implicit

更简洁的设计将明确允许指定一个组：

with TimerGroup('Doing big task') as big_task_tg:
    with Timer('Foo', big_task_tg):
      foo_result = foo()
    with Timer('Bar', big_task_tg):
      bar(baz(foo_result))

另一方面，您始终可以使用 traceback.extract_stack 并在上游查找特定函数的调用。它对于日志记录和错误报告非常有用，并且对于确保特定函数仅在特定上下文中被调用具有一定的帮助。但它往往会产生很难跟踪的依赖关系。

尽管您可以尝试，但我会避免将其用于分组计时器。如果您非常需要自动分组，@Martijn-Pieters 的方法要好得多。

Answer 3

如果您需要做的只是根据您在其中执行的嵌套上下文管理器的数量来调整缩进级别，那么有一个名为 indent_level 的 class 属性，并在每次执行时进行调整进入和退出上下文管理器。类似于以下内容：

class Context:
    indent_level = 0

    def __init__(self, name):
        self.name = name

    def __enter__(self):
        print(' '*4*self.indent_level + 'Entering ' + self.name)
        self.adjust_indent_level(1)
        return self

    def __exit__(self, *a, **k):
        self.adjust_indent_level(-1)
        print(' '*4*self.indent_level + 'Exiting ' + self.name)

    @classmethod
    def adjust_indent_level(cls, val):
        cls.indent_level += val

并将其用作：

>>> with Context('Outer') as outer_context:
        with Context('Inner') as inner_context:
            print(' '*inner_context.indent_level*4 + 'In the inner context')


Entering Outer
    Entering Inner
        In the inner context
    Exiting Inner
Exiting Outer

检测上下文管理器嵌套

Detecting context manager nesting

python

nested

with-statement

contextmanager