Python 如何记录某个函数执行期间调用的其他函数?
可以在每次节点或边转换开始前启用 sys.setprofile(),转换结束后关闭,只记录属于 ConversionFunctions 的 Python 函数调用。这样可以捕获函数之间的内部调用,也无需包装原函数,不会改变原函数签名。
如果仍想使用装饰器,可以配合 functools.wraps 和 inspect.signature() 处理签名检查。
使用 sys.setprofile 记录一次转换中的调用
sys.setprofile() 可以接收 Python 函数的 call、return 等事件。在 call 事件中,frame.f_code 是当前进入的函数的代码对象,可以用它判断该函数是否属于需要记录的范围。
下面的示例先收集 ConversionFunctions 暴露的 Python 函数,再记录一次转换中的调用。它适用于将函数放在模块或类上的常见写法,也能处理通过属性访问得到的绑定方法。
import inspect
import sys
def collect_function_codes(library):
codes = set()
for _, member in inspect.getmembers(library):
if inspect.ismethod(member):
member = member.__func__
if inspect.isfunction(member):
codes.add(member.__code__)
return codes
# 如果之后会动态添加或替换函数,需要重新收集。
conversion_codes = collect_function_codes(ConversionFunctions)
def convert_node(node_info):
node_conversion_fxn = getattr(
ConversionFunctions, node_info['fxn_name']
)
arguments = node_info['arguments']
called_functions = []
def profile(frame, event, arg):
if event == 'call' and frame.f_code in conversion_codes:
called_functions.append(frame.f_code.co_name)
previous_profile = sys.getprofile()
if previous_profile is not None:
raise RuntimeError('当前线程已经安装了 profile 回调')
sys.setprofile(profile)
try:
result = node_conversion_fxn(**arguments)
finally:
sys.setprofile(previous_profile)
return result, called_functions
每次调用都会创建独立的 called_functions,可以直接将记录关联到当前节点或边:
result, called_functions = convert_node(node_info)
假设入口函数 convert_a 内部先后调用了 normalize、convert_b 和 normalize,并且这些函数都在收集范围内,记录顺序就是:
['convert_a', 'normalize', 'convert_b', 'normalize']
记录中保留了重复调用。如果只需要知道哪些函数被使用过,可以在转换结束后按首次出现的顺序去重:
used_functions = list(dict.fromkeys(called_functions))
明确函数筛选的范围
示例按代码对象筛选,只记录从 ConversionFunctions 的成员中收集到的函数。如果这些函数还会调用其他模块中的辅助函数,而你也想记录它们,需要把对应函数的代码对象加入集合。
如果 ConversionFunctions 明确是一个模块,且目标是记录所有在该模块中定义的 Python 函数,也可以按定义函数时所属的全局命名空间筛选:
library_globals = vars(ConversionFunctions)
def profile(frame, event, arg):
if event == 'call' and frame.f_globals is library_globals:
called_functions.append(frame.f_code.co_name)
这两种筛选方式的范围不同:代码对象集合由你明确选定函数;模块全局命名空间筛选则覆盖在该模块中定义的函数,包括未作为模块顶层成员暴露的内部函数。后者不会自动包含从其他模块导入的函数。
如果存在同名函数,单独记录 co_name 无法区分它们,可以同时保存位置:
code = frame.f_code
called_functions.append(
(code.co_filename, code.co_firstlineno, code.co_name)
)
装饰器如何配合函数签名检查
如果采用装饰器,可以使用 functools.wraps 包装原函数,并通过 inspect.signature() 获取签名。wraps 会设置 __wrapped__,而 inspect.signature() 默认会沿着它查找原函数。
import inspect
from functools import wraps
def tracked(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 在这里记录调用;实际使用时应接入当前转换的记录容器。
return func(*args, **kwargs)
return wrapper
@tracked
def convert_value(value, scale=1):
return value * scale
arguments = {'value': 3, 'scale': 2}
signature = inspect.signature(convert_value)
signature.bind(**arguments)
result = convert_value(**arguments)
signature.bind(**arguments) 会检查参数能否按照该签名绑定,参数不匹配时抛出 TypeError。它不会检查参数值是否符合类型注解,也不会执行函数。
wraps 并没有把包装函数本身的参数声明改成原函数的参数声明。如果现有检查直接读取 __code__,或者使用 inspect.signature(..., follow_wrapped=False),仍然会看到包装层的信息。
使用时的边界
- 上面的实现面向当前线程内同步执行的普通 Python 函数。新线程中的调用需要另行设置记录机制。
- 对于生成器或
async def函数,仅创建生成器或协程对象不会执行完整函数体;记录范围需要覆盖实际迭代或等待执行的阶段。暂停和恢复也需要单独考虑。 call事件用于 Python 函数。C 扩展函数有不同的 profile 事件,示例没有记录它们。sys.setprofile()会增加运行开销。即使最终只保存目标库的函数,回调仍会接收到当前线程中其他函数的相关事件。- 示例检测到已有 profile 回调时会报错,避免覆盖其他性能分析工具的回调。转换期间发生异常时,
finally仍会恢复原来的设置,但正常返回的记录列表不会交付给调用方。如果需要保留失败节点的记录,可以把列表传入或交给外层统一保存。
备注:内容仅供参考。
