我们如何将Python上下文管理器“关联”到出现在其块中的变量？

Question

据我所知，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)似乎以某种方式访问了“当前”上下文管理器，因此它可以在幕后将每条新创建的语句关联到它。但是，库如何访问上下文管理器呢？

Answer 1

PyMC3通过在线程局部变量类中将线程局部变量维护为类变量来实现这一点。Model是从Context继承的。

每次在模型上调用with时，都会将当前模型推送到特定于线程的上下文堆栈上。因此，堆栈的顶部总是引用作为上下文管理器的最内部(最近的)模型。

Contexts (因此是Models)有一个.get_context() 类方法来获取上下文堆栈的顶部。

当创建Distribution以将自己与最内部的模型关联时，它们会调用Model.get_context()。

因此，简而言之：

1. with model将model推到上下文堆栈上。这意味着with块、type(model).contexts或Model.contexts或Context.contexts中现在包含了model作为它的最后一个(最顶层)元素。
2. Distribution.__init__()调用Model.get_context() (note M)，它返回上下文堆栈的顶部。在我们的例子中，这是model。上下文堆栈是线程本地的(每个线程有一个)，但它不是特定于实例的。如果只有一个线程，那么不管模型的数量如何，也只有一个上下文堆栈。
3. 退出上下文管理器时。model从上下文堆栈中弹出。

Answer 2

我不知道在这种特殊情况下它是如何工作的，但是通常您会使用一些“幕后魔法”：

class Parent:
    def __init__(self):
        self.active_child = None

    def ContextManager(self):
        return Child(self)

    def Attribute(self):
        return self.active_child.Attribute()

class Child:
    def __init__(self,parent):
        self.parent = parent

    def __enter__(self):
        self.parent.active_child = self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.parent.active_child = None

    def Attribute(self):
        print("Called Attribute of child")

使用此代码：

p = Parent()
with p.ContextManager():
    attr = p.Attribute()

将产生以下产出：

Called Attribute of child

Answer 3

在输入和退出上下文管理器块时，还可以检查堆栈中是否存在局部变量()变量，并确定哪些变量已经更改。

class VariablePostProcessor(object):
    """Context manager that applies a function to all newly defined variables in the context manager.

    with VariablePostProcessor(print):
        a = 1
        b = 3

    It uses the (name, id(obj)) of the variable & object to detect if a variable has been added.
    If a name is already binded before the block to an object, it will detect the assignment to this name
    in the context manager block only if the id of the object has changed.

    a = 1
    b = 2
    with VariablePostProcessor(print):
        a = 1
        b = 3
    # will only detect 'b' has newly defined variable/object. 'a' will not be detected as it points to the
    # same object 1
    """

    @staticmethod
    def variables():
        # get the locals 2 stack above
        # (0 is this function, 1 is the __init__/__exit__ level, 2 is the context manager level)
        return {(k, id(v)): v for k, v in inspect.stack()[2].frame.f_locals.items()}

    def __init__(self, post_process):
        self.post_process = post_process
        # save the current stack
        self.dct = self.variables()

    def __enter__(self):
        return

    def __exit__(self, type, value, traceback):
        # compare variables defined at __exist__ with variables defined at __enter__
        dct_exit, dct_enter = self.variables(), self.dct
        for (name, id_) in set(dct_exit).difference(dct_enter):
            self.post_process(name, dct_exit[(name, id_)])

典型的用途可以是：

# let us define a Variable object that has a 'name' attribute that can be defined at initialisation time or later
class Variable:
    def __init__(self, name=None):
        self.name = name

# the following code
x = Variable('x')
y = Variable('y')
print(x.name, y.name)

# can be replaced by
with VariablePostProcessor(lambda name, obj: setattr(obj, "name", name)):
    x = Variable()
    y = Variable()
print(x.name, y.name)

# in such case, you can also define as a convenience
import functools
AutoRenamer = functools.partial(VariablePostProcessor, post_process=lambda name, obj: setattr(obj, "name", name))

# and rewrite the above code as
with AutoRenamer():
    x = Variable()
    y = Variable()
print(x.name, y.name)  # => x y