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

如何为抽象方法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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.10.05 23:48:00