优雅的Python解决方案，用于跟踪/记录经过的时间？

Question

为了在日志中捕获有意义的经过时间信息，我在多个函数中复制了以下时间捕获和日志记录代码：

import time
import datetime

def elapsed_str(seconds):
    """ Returns elapsed number of seconds in format '(elapsed HH:MM:SS)' """
    return "({} elapsed)".format(str(datetime.timedelta(seconds=int(seconds))))

def big_job(job_obj):
    """ Do a big job and return the result """
    start = time.time()
    logging.info(f"Starting big '{job_obj.name}' job...")
    logging.info(f"Doing stuff related to '{job_type}'...")
    time.sleep(10)  # Do some stuff...
    logging.info(f"Big '{job_obj.name}' job completed! "
                 f"{elapsed_str(time.time() - start)}")
    return my_result

使用示例输出：

big_job("sheep-counting")
# Log Output:
#   2019-09-04 01:10:48,027 - INFO - Starting big 'sheep-counting' job...
#   2019-09-04 01:10:48,092 - INFO - Doing stuff related to 'sheep-counting'
#   2019-09-04 01:10:58,802 - INFO - Big 'sheep-counting' job completed! (0:00:10 elapsed)

我正在寻找一种优雅的(pythonic)方法，将这些多余的行从每次重写的代码中删除：

1. start = time.time() -应该会自动捕获函数启动时的启动时间。
2. time.time() - start应该使用以前捕获的开始时间，并从now()推断当前时间。(理想情况下，elapsed_str()可以使用零参数进行调用。)

我的具体用例是生成数据科学/数据工程领域中的大型数据集。运行时可以是从几秒钟到几天之间的任何地方，至关重要的是：(1)日志可以很容易地搜索(在本例中是“已过”一词)和(2)添加日志的开发人员成本非常低(因为我们无法提前知道哪些作业可能比较慢，而且一旦发现性能问题，我们可能无法修改源代码)。

Answer 1

推荐的方法是从3.7开始使用time.perf_counter()和time.perf_counter_ns()。

为了度量函数的运行时，使用装饰器是很舒服的。例如，以下内容：

import time

def benchmark(fn):
    def _timing(*a, **kw):
        st = time.perf_counter()
        r = fn(*a, **kw)
        print(f"{fn.__name__} execution: {time.perf_counter() - st} seconds")
        return r

    return _timing

@benchmark
def your_test():
    print("IN")
    time.sleep(1)
    print("OUT")

your_test()

(c)这个装饰器的代码是从 package稍微修改的

Answer 2

如果我正确地理解了你，你可以写一个装饰器来计时这个函数。

这里有一个很好的例子：https://stackoverflow.com/a/5478448/6001492

Answer 3

对于其他人的用例来说，这可能有些过分，但我发现的解决方案需要一些困难的处理，我将在这里为任何想要完成类似任务的人记录它们。

1.用于动态计算f-字符串的助手函数。

def fstr(fstring_text, locals, globals=None):
    """
    Dynamically evaluate the provided fstring_text

    Sample usage:
        format_str = "{i}*{i}={i*i}"
        i = 2
        fstr(format_str, locals()) # "2*2=4"
        i = 4
        fstr(format_str, locals()) # "4*4=16"
        fstr(format_str, {"i": 12}) # "10*10=100"
    """
    locals = locals or {}
    globals = globals or {}
    ret_val = eval(f'f"{fstring_text}"', locals, globals)
    return ret_val

2. @logged装饰器类

class logged(object):
    """
    Decorator class for logging function start, completion, and elapsed time.
    """

    def __init__(
        self,
        desc_text="'{desc_detail}' call to {fn.__name__}()",
        desc_detail="",
        start_msg="Beginning {desc_text}...",
        success_msg="Completed {desc_text}  {elapsed}",
        log_fn=logging.info,
        **addl_kwargs,
    ):
        """ All arguments optional """
        self.context = addl_kwargs.copy()  # start with addl. args
        self.context.update(locals())  # merge all constructor args
        self.context["elapsed"] = None
        self.context["start"] = time.time()

    def re_eval(self, context_key: str):
        """ Evaluate the f-string in self.context[context_key], store back the result """
        self.context[context_key] = fstr(self.context[context_key], locals=self.context)

    def elapsed_str(self):
        """ Return a formatted string, e.g. '(HH:MM:SS elapsed)' """
        seconds = time.time() - self.context["start"]
        return "({} elapsed)".format(str(datetime.timedelta(seconds=int(seconds))))

    def __call__(self, fn):
        """ Call the decorated function """

        def wrapped_fn(*args, **kwargs):
            """
            The decorated function definition. Note that the log needs access to 
            all passed arguments to the decorator, as well as all of the function's
            native args in a dictionary, even if args are not provided by keyword.
            If start_msg is None or success_msg is None, those log entries are skipped.
            """
            self.context["fn"] = fn
            fn_arg_names = inspect.getfullargspec(fn).args
            for x, arg_value in enumerate(args, 0):
                self.context[fn_arg_names[x]] = arg_value
            self.context.update(kwargs)
            desc_detail_fn = None
            log_fn = self.context["log_fn"]
            # If desc_detail is callable, evaluate dynamically (both before and after)
            if callable(self.context["desc_detail"]):
                desc_detail_fn = self.context["desc_detail"]
                self.context["desc_detail"] = desc_detail_fn()

            # Re-evaluate any decorator args which are fstrings
            self.re_eval("desc_detail")
            self.re_eval("desc_text")
            # Remove 'desc_detail' if blank or unused
            self.context["desc_text"] = self.context["desc_text"].replace("'' ", "")
            self.re_eval("start_msg")
            if self.context["start_msg"]:
                # log the start of execution
                log_fn(self.context["start_msg"])
            ret_val = fn(*args, **kwargs)
            if desc_detail_fn:
                # If desc_detail callable, then reevaluate
                self.context["desc_detail"] = desc_detail_fn()
            self.context["elapsed"] = self.elapsed_str()
            # log the end of execution
            log_fn(fstr(self.context["success_msg"], locals=self.context))
            return ret_val

        return wrapped_fn

样本使用情况：

@logged()
def my_func_a():
    pass
# 2019-08-18 - INFO - Beginning call to my_func_a()...
# 2019-08-18 - INFO - Completed call to my_func_a()  (00:00:00 elapsed)


@logged(log_fn=logging.debug)
def my_func_b():
    pass
# 2019-08-18 - DEBUG - Beginning call to my_func_b()...
# 2019-08-18 - DEBUG - Completed call to my_func_b()  (00:00:00 elapsed)


@logged("doing a thing")
def my_func_c():
    pass
# 2019-08-18 - INFO - Beginning doing a thing...
# 2019-08-18 - INFO - Completed doing a thing  (00:00:00 elapsed)


@logged("doing a thing with {foo_obj.name}")
def my_func_d(foo_obj):
    pass
# 2019-08-18 - INFO - Beginning doing a thing with Foo...
# 2019-08-18 - INFO - Completed doing a thing with Foo  (00:00:00 elapsed)


@logged("doing a thing with '{custom_kwarg}'", custom_kwarg="foo")
def my_func_e(foo_obj):
    pass
# 2019-08-18 - INFO - Beginning doing a thing with 'foo'...
# 2019-08-18 - INFO - Completed doing a thing with 'foo'  (00:00:00 elapsed)

结论

与简单的装饰解决方案相比，主要的优点是：

1. 通过利用f-字符串的延迟执行，以及从装饰器构造函数和ass函数调用本身注入上下文变量，日志消息传递可以很容易地定制为人类可读的。
2. (最重要的是)，几乎任何函数参数的派生都可以用于区分后续调用中的日志-而不需要更改函数本身的定义方式。
3. 高级回调场景可以通过向装饰器参数desc_detail发送函数或复杂对象来实现，在这种情况下，函数将在函数执行之前和之后得到评估。这最终可以扩展为使用回调函数来计算创建的数据表中的行数(例如)，并将表行计数包括在完成日志中。