Python：在代码块的每一行添加代码例程

Question

我希望能在另一个代码块的每一行运行一段代码。例如，想要在执行函数的下一行之前或之后评估全局变量。

例如，在下面的示例中，我尝试在foo()函数的每一行之前打印“hello”。我认为一个装饰器可以帮助我，但它需要一些内省特性来编辑foo()函数的每一行并在其之前或之后添加我想要的内容。

我正在尝试执行类似于此的操作：

>>> def foo():
...    print 'bar'
...    print 'barbar'
...    print 'barbarbar'

>>> foo()
hello
bar
hello
barbar 
hello
barbarbar

我该如何执行这个操作？__code__对象是否有帮助？我需要同时使用装饰器和内省吗？

编辑：以下是此线程目标的另一个示例：

>>> def foo():
...    for i in range(0,3):
...        print 'bar'

>>> foo()
hello
bar
hello
bar 
hello
bar

在打印每个“bar”之前，我希望打印一个“hello”。主要目的是在执行代码的下一行之前能够执行另一个函数或测试任何类型的全局变量。想象一下，如果全局变量为True，则代码继续执行下一行；而如果全局变量为False，则停止函数执行。

编辑：某种程度上，我正在寻找一种将代码注入另一个代码块的工具。

编辑：感谢unutbu，我已经实现了这段代码：

import sys
import time
import threading

class SetTrace(object):
    """
    with SetTrace(monitor):
    """
    def __init__(self, func):
        self.func = func
    def __enter__(self):
        sys.settrace(self.func)
        return self
    def __exit__(self, ext_type, exc_value, traceback):
        sys.settrace(None)
        # http://effbot.org/zone/python-with-statement.htm
        # When __exit__ returns True, the exception is swallowed.
        # When __exit__ returns False, the exception is reraised.
        # This catches Sentinel, and lets other errors through
        # return isinstance(exc_value, Exception)

def monitor(frame, event, arg):
    if event == "line":
        if not running:
            raise Exception("global running is False, exiting")
    return monitor

def isRunning(function):
    def defaultBehavior(*args):
        with SetTrace(monitor):
            ret = function(*args)
            return ret
    return defaultBehavior

@isRunning
def foo():
    while True:
        time.sleep(1)
        print 'bar'

global running
running = True
thread = threading.Thread(target = foo)
thread.start()
time.sleep(3)
running = False

- afiah

Answer 1

也许您正在寻找 sys.settrace ：

import sys
class SetTrace(object):
    def __init__(self, func):
        self.func = func

    def __enter__(self):
        sys.settrace(self.func)
        return self

    def __exit__(self, ext_type, exc_value, traceback):
        sys.settrace(None)

def monitor(frame, event, arg):
    if event == "line":
        print('hello')
        # print(frame.f_globals) 
        # print(frame.f_locals)  
    return monitor



def foo():
   print 'bar'
   print 'barbar'
   print 'barbarbar'

with SetTrace(monitor):
    foo()

产量

hello
bar
hello
barbar
hello
barbarbar
hello

在monitor内，您可以使用frame.f_locals和frame.f_globals访问foo的本地变量和全局变量。

请参阅此文章，了解如何使用sys.settrace进行调试的示例。

---

如何在monitor中停止foo：

最优雅的方法是在foo内部放置一个条件语句，以便foo检查何时退出。然后，您可以从monitor内部操纵条件的值来控制foo的退出时间。

但是，如果您不想或无法更改foo，则另一种方法是从monitor中引发异常。异常将通过帧堆栈向上冒泡，直到被捕获。如果您在SetTrace.__exit__中捕获它，则控制流将继续进行，就像foo刚刚已退出一样。

import sys
class Sentinel(Exception): pass

class SetTrace(object):
    """
    with SetTrace(monitor):
        ...
    """
    def __init__(self, func):
        self.func = func

    def __enter__(self):
        sys.settrace(self.func)
        return self

    def __exit__(self, ext_type, exc_value, traceback):
        sys.settrace(None)
        # http://effbot.org/zone/python-with-statement.htm
        # When __exit__ returns True, the exception is swallowed.
        # When __exit__ returns False, the exception is reraised.

        # This catches Sentinel, and lets other errors through
        return isinstance(exc_value, Sentinel)

def monitor(frame, event, arg):
    if event == "line":
        l = frame.f_locals
        x = l.get('x', 0)
        print('x = {}'.format(x))
        if x > 3:
            raise Sentinel()
    return monitor

def foo():
    x = 0
    while True:
        print 'bar'
        x += 1

with SetTrace(monitor):
    foo()

- unutbu

---

非常感谢，这似乎与你说的很接近。线程的主要目的不是调试，而是通过改变另一个全局变量的值来停止正在执行的代码块，然后该函数在执行每一行代码之前会评估该变量。这将在代码的每一行中添加一个抢占点，如果全局开关为假，则函数停止而不继续执行到最后；如果开关为真，则函数执行该行并继续到下一行。我表达清楚了吗？ - afiah

是的，您可以在monitor中检查和更改局部变量和全局变量的值。 - unutbu

谢谢，但我能停止当前被监视函数的执行吗？在这种情况下，可以停止执行foo()吗？（我是否可以在需要时添加返回行以在中间完成我的函数？） - afiah

我已经编辑了上面的帖子，展示了如何在monitor内部终止foo。 - unutbu

非常感谢，它确实有效，我将在我的帖子中添加个人修改。 - afiah

Answer 2

听起来你需要一个调试器，请看一下内置的pdb。使用pdb，你可以做到这些：

>>> def foo():
...     import pdb;pdb.set_trace()
...     print 'bar'
...     print 'barbar'
...     print 'barbarbar'
... 
>>> 
>>> foo()
> <stdin>(3)foo()
(Pdb) print 'hi'
hi
(Pdb) n
bar
> <stdin>(4)foo()
(Pdb) n
barbar
> <stdin>(5)foo()
(Pdb) n
barbarbar
--Return--
> <stdin>(5)foo()->None
(Pdb) n
--Return--
> <stdin>(1)<module>()->None
(Pdb) n
>>> 

像大多数其他调试器一样，这个调试器允许你逐行查看代码。你可以查阅文档以获取更多信息，但在上面的例子中，pdb.set_trace() -call 设置了调试入口点并打开了pdb控制台。从控制台中，你可以修改变量和进行各种操作。n只是next的简写，表示向前执行一步。

- msvalkon

Answer 3

最佳答案可能取决于您实际想要做什么，但为了完成您所要求的操作，我会这样做：

from itertools import chain

def print_(s):
    print s

def print_f(s):
   return (lambda: print_(s))

def execute_several(functions):
    for f in functions:
        f()

def prepend_printhello(function):
    return (print_f("hello"), function)

def foo():
    execute_several(chain(*map(prepend_printhello, map(print_f, ("bar", "barbar", "barbarbar")))))

- rlms

---

这与我正在寻找的不符，因为它会更改编写foo()函数的方式并创建3个lambda函数。我正在寻找一种保持传统编码方式并添加一个装饰器来执行所有这些工作的方法。 - afiah

顺便提一下：有些行为很难处理，比如想象一个带有for循环的foo()函数，我想在每个循环周期中执行我的打印“hello”的操作。换句话说，我正在寻找代码每行的抢占点。 - afiah