You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python函数能否引用自身或其他函数中已记录的日志事件?

Python函数能否引用自身或其他函数作用域中已记录的日志事件?

示例:
在一个大型程序中,process_files(files)处理20个文件时,有4个文件触发并处理了OSError,通过logging.error(f'foo exception handled for {file} due to {bar} error')记录日志,其余文件处理成功。后续progress_report()需要告知用户其他函数中发生的ERROR级及以上事件。

已考虑的方案:
创建自定义Error_Logger类并设为全局变量my_logger = Error_Logger(),在各函数中用它收集记录错误,其他函数可查询errors或member_functions属性获取错误信息,代码如下:

import logging
import inspect

class Error_Logger:
    def __init__(self):
        self.errors = {}
        
    @property
    def member_functions(self):
        return self.errors.keys()
    
    def log_error(self, error, level='WARNING', data=None):
        caller_name = self._caller_name()
        num_level = logging.getLevelName(level)
        if not isinstance(num_level, int):
            num_level = 0
        level_name = logging.getLevelName(num_level)
        logger = getattr(logging, str(level_name.lower()))
        logger(f'{caller_name}: {error}')
        
        if caller_name not in self.errors:
            self.errors[caller_name] = {}
        
        self.errors[caller_name][len(self.errors[caller_name])] = {
            'error': error,
            'level': num_level,
            'data': data
        }
    
    @staticmethod
    def _error_set(errors):
        error_set = set()
        for k, v in errors.items():
            error_level = v.get('level', None)
            if error_level:
                error_set.add(error_level)
        return error_set
    
    def max_error(self, function=None):
        if function:
            errors = self.errors.get(function, None)
            max_error = max(self._error_set(errors)) if errors else None
        else:
            all_errors = set()
            for v in self.errors.values():
                all_errors.update(self._error_set(v))
            max_error = max(all_errors) if all_errors else None
        return max_error
    
    def _caller_name(self, level=2):
        return inspect.stack()[level][3]

my_logger = Error_Logger()

def process_files(files):
    processed_files = []
    try:
        for f in files:
            # 模拟文件处理逻辑
            if f == 'b':
                raise OSError("Permission denied")
            processed_files.append(f)
    except Exception as e:
        my_logger.log_error(f'bad things happened {e}', 'ERROR', e)

    return processed_files

def progress_report():
    if 'process_files' in my_logger.errors and len(my_logger.errors['process_files']):
        print(f'bad things happened with {len(my_logger.errors["process_files"])} files')

files = ['a', 'b', 'c']
foo = process_files(files)
progress_report()

该方案的不足:

  • 存在重复造轮子的问题,应该有更成熟的现成解决方案
  • 全局变量需要提前初始化,且始终占用全局命名空间
  • 全局变量会增加未来导出模块中函数的难度
  • 全局变量容易引发难以排查的意外变更

内容的提问来源于stack exchange,提问作者Aaron Ciuffo

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.30 10:30:38