如何为抽象方法calculate_insights内调用的所有函数自动添加计时日志
方案1:模板方法+实例方法临时装饰(推荐,低侵入高性能)
该方案仅需修改父类Insights的代码,子类业务逻辑无需做任何调整,完全满足你不想新增冗余日志代码的需求。
核心思路是用模板方法模式封装执行流程:将原抽象方法重命名为内部抽象实现,父类实现公开的执行入口,在执行业务逻辑前给db、bq实例的方法统一挂载计时装饰器,执行完成后自动复原原始方法,不会影响其他地方对这两个实例的调用。
from abc import ABC, abstractmethod import time import logging from functools import wraps # 通用计时装饰器,统一维护日志逻辑 def time_logger(func): @wraps(func) def wrapper(*args, **kwargs): start_time = time.perf_counter() try: return func(*args, **kwargs) finally: cost = time.perf_counter() - start_time logging.info(f"方法[{func.__name__}]执行耗时:{cost:.4f}s") return wrapper class Insights(ABC): def __init__(self): self.bq = BigQueryLayer() self.db = DatabaseLayer() # 存储原始方法,避免重复装饰 self._origin_db_methods = {} self._origin_bq_methods = {} def _add_time_log(self, instance, origin_store: dict): """给实例的所有公共业务方法加计时装饰""" for attr_name in dir(instance): # 过滤私有方法、内置方法,仅装饰业务方法 if not attr_name.startswith('_') and callable(getattr(instance, attr_name)): origin_func = getattr(instance, attr_name) origin_store[attr_name] = origin_func setattr(instance, attr_name, time_logger(origin_func)) def _recover_origin_method(self, instance, origin_store: dict): """复原实例的原始方法,避免影响其他业务场景调用""" for attr_name, origin_func in origin_store.items(): setattr(instance, attr_name, origin_func) origin_store.clear() @abstractmethod def _calculate_insights(self): # 子类原calculate_insights的逻辑迁移到此处即可,内部代码无需修改 pass def calculate_insights(self): # 执行前挂载计时装饰 self._add_time_log(self.db, self._origin_db_methods) self._add_time_log(self.bq, self._origin_bq_methods) try: return self._calculate_insights() finally: # 执行完成后复原方法 self._recover_origin_method(self.db, self._origin_db_methods) self._recover_origin_method(self.bq, self._origin_bq_methods) # 子类仅需将原方法名改为_calculate_insights,内部业务代码一行不用动 class BrandInsights(Insights): def _calculate_insights(self): self.db.extend_customer_loyalty() self.db.extend_brand_combiners() self.db.extend_brand_recency() ... class StoreInsights(Insights): def _calculate_insights(self): self.db.extend_competition_view() self.db.extend_busiest_hours() ...
方案2:系统调用追踪(适合调试/全量统计场景,无需修改子类代码)
如果你连子类的方法名都不想改,可以用sys.settrace实现函数调用的全局追踪,仅在执行calculate_insights期间生效,自动统计所有调用函数的耗时。该方案性能开销比方案1高,适合测试、压测场景使用。
import sys import time import logging from abc import ABC, abstractmethod class Insights(ABC): def __init__(self): self.bq = BigQueryLayer() self.db = DatabaseLayer() self._trace_running = False self._func_start_time = {} def _call_trace(self, frame, event, arg): if not self._trace_running: return if event == 'call': func_name = frame.f_code.co_name if not func_name.startswith('_'): self._func_start_time[func_name] = time.perf_counter() elif event == 'return': func_name = frame.f_code.co_name if func_name in self._func_start_time: cost = time.perf_counter() - self._func_start_time.pop(func_name) logging.info(f"方法[{func_name}]执行耗时:{cost:.4f}s") return self._call_trace @abstractmethod def calculate_insights(self): pass def run_with_timing(self): self._trace_running = True sys.settrace(self._call_trace) try: return self.calculate_insights() finally: sys.settrace(None) self._trace_running = False self._func_start_time.clear()
使用时直接调用run_with_timing()方法即可,子类calculate_insights的代码完全不需要调整。
内容的提问来源于stack exchange,提问作者Knoerifast
相关产品推荐
相关产品推荐

