当你需要测量函数执行时间并记录日志时,一个实用的工具是Timer类。本文将介绍如何使用Timer类来装饰同步和异步函数,并展示它们的用法。
首先,让我们来看一下Timer类的定义:
import asyncio
import time
from functools import wraps
from loguru import logger
logger.add(sys.stdout, level='DEBUG'
format='[<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green>] '
'[<magenta>{process.name}</magenta>:<yellow>{thread.name}</yellow>] '
'[<cyan>{name}</cyan>:<cyan>{function}</cyan>:<yellow>{line}</yellow>] '
'[<level>{level}</level>] '
'<level>{message}</level>',
)
class Timer:
def __init__(self, msg: str or int = None):
""" Timer类用于测量函数执行时间和记录日志。
:param msg: 日志信息的描述
:type msg: str or int
"""
self.msg = msg
def __call__(self, func):
""" __call__方法使Timer类的实例可以像函数一样被调用。
:param func: 被装饰的函数
:type func: function
:return: 装饰后的函数
"""
if asyncio.iscoroutinefunction(func):
return self._async_wrapper(func)
else:
return self._wrapper(func)
def _wrapper(self, func):
""" 同步函数的装饰器。
:param func: 被装饰的同步函数
:type func: function
:return: 装饰后的同步函数
"""
@wraps(func)
def wrapper(*args, **kwargs):
time_start = time.time()
res = func(*args, **kwargs)
self._log_execution_time(func, time_start)
return res
return wrapper
def _async_wrapper(self, func):
""" 异步函数的装饰器。
:param func: 被装饰的异步函数
:type func: function
:return: 装饰后的异步函数
"""
@wraps(func)
async def async_wrapper(*args, **kwargs):
time_start = time.time()
res = await func(*args, **kwargs)
self._log_execution_time(func, time_start)
return res
return async_wrapper
def _log_execution_time(self, func, start_time):
""" 记录函数执行时间的日志。
:param func: 函数对象
:type func: function
:param start_time: 函数开始执行的时间戳
:type start_time: float
"""
end_time = time.time()
execution_time = end_time - start_time
module_name = func.__module__
function_name = func.__name__
message = f"{self.msg}, " if self.msg else ""
log_message = f"{message}运行时间:{execution_time:.5f}秒"
logger.patch(lambda r: r.update(function=f"{r.get('function')}({module_name}:{function_name})")).info(
log_message)
Timer类包含了几个核心方法,其中包括了装饰器函数__call__、同步函数装饰器_wrapper和异步函数装饰器_async_wrapper。这些方法使得Timer类可以灵活地装饰不同类型的函数,并测量它们的执行时间。
接下来,我们来看一下Timer类的用法示例:
from timer import Timer
# 定义一个同步函数并使用Timer装饰器
@Timer("执行任务")
def sync_task():
# 模拟一个耗时任务
time.sleep(2)
print("同步任务完成")
# 定义一个异步函数并使用Timer装饰器
@Timer("执行任务")
async def async_task():
# 模拟一个耗时任务
await asyncio.sleep(2)
print("异步任务完成")
# 调用同步函数
sync_task()
# 调用异步函数
asyncio.run(async_task())
在上面的示例中,我们首先创建了一个Timer实例,并传入了一个描述信息"执行任务"。然后,我们分别定义了一个同步函数sync_task和一个异步函数async_task,并使用Timer实例对它们进行装饰。最后,我们分别调用了这两个函数,观察它们的执行时间和日志记录情况。
通过Timer类,我们可以方便地测量函数的执行时间,并将执行时间记录在日志中,从而更好地了解程序的性能表现。
发表评论
共 0 条评论
暂无评论