Python抽象基类子类方法执行时自动实现耗时统计的方法
实现方案
直接替换原有Encoder基类代码即可,全程不需要修改任何子类实现,所有继承该基类的编码器会自动获得编解码耗时统计能力。
推荐使用Python 3.6+原生支持的__init_subclass__钩子实现,侵入性最低:
import abc import time from functools import wraps from typing import Tuple # 项目内已有的DataMapper、MetadataCollector类型保持原有导入逻辑即可 class Encoder(abc.ABC): def __init__(self): self._encode_time = [] self._decode_time = [] def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs) # 自动包装子类实现的encode方法 original_encode = cls.encode @wraps(original_encode) def timed_encode(self, decoded_data, **kwargs): start = time.perf_counter() try: return original_encode(self, decoded_data, **kwargs) finally: self._encode_time.append(time.perf_counter() - start) cls.encode = timed_encode # 自动包装子类实现的decode方法 original_decode = cls.decode @wraps(original_decode) def timed_decode(self, encoded_data, metadata, **kwargs): start = time.perf_counter() try: return original_decode(self, encoded_data, metadata, **kwargs) finally: self._decode_time.append(time.perf_counter() - start) cls.decode = timed_decode @abc.abstractmethod def encode(self, decoded_data: DataMapper, **kwargs) -> Tuple[DataMapper, MetadataCollector]: pass @abc.abstractmethod def decode(self, encoded_data: DataMapper, metadata: MetadataCollector, **kwargs) -> DataMapper: pass
实现说明
- 逻辑对业务代码完全透明:子类原有入参、返回值、异常抛出逻辑完全不受影响,不需要做任何适配
- 用
try/finally块保证即使编解码过程抛出异常,本次操作的耗时也会被正常记录,不会出现统计漏项 - 用
functools.wraps保留原有方法的签名、文档字符串等元信息,不会破坏依赖方法元信息的框架校验逻辑 - 替换了示例中已在Python3.8版本移除的
time.clock(),改用官方推荐的time.perf_counter()做耗时统计,精度更高、不受系统时钟调整影响 - 统计结果依然存储在实例的
_encode_time、_decode_time列表中,原有读取统计数据的逻辑不需要做任何调整
如果项目运行在Python3.6以下版本(不支持__init_subclass__),可以换用元类实现完全相同的效果:
import abc import time from functools import wraps from typing import Tuple class EncoderMeta(abc.ABCMeta): def __init__(cls, name, bases, namespace): super().__init__(name, bases, namespace) # 跳过基类本身,只包装具体子类的实现 if name != 'Encoder' and not cls.__abstractmethods__: original_encode = cls.encode @wraps(original_encode) def timed_encode(self, decoded_data, **kwargs): start = time.perf_counter() try: return original_encode(self, decoded_data, **kwargs) finally: self._encode_time.append(time.perf_counter() - start) cls.encode = timed_encode original_decode = cls.decode @wraps(original_decode) def timed_decode(self, encoded_data, metadata, **kwargs): start = time.perf_counter() try: return original_decode(self, encoded_data, metadata, **kwargs) finally: self._decode_time.append(time.perf_counter() - start) cls.decode = timed_decode class Encoder(metaclass=EncoderMeta): def __init__(self): self._encode_time = [] self._decode_time = [] @abc.abstractmethod def encode(self, decoded_data: DataMapper, **kwargs) -> Tuple[DataMapper, MetadataCollector]: pass @abc.abstractmethod def decode(self, encoded_data: DataMapper, metadata: MetadataCollector, **kwargs) -> DataMapper: pass
内容的提问来源于stack exchange,提问作者graille
相关产品推荐
相关产品推荐

