Reputation: 2571
With the goal of capturing meaningful elapsed-time information in for logs, I have replicated the following time-capture and logging code across many functions:
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
With sample usage output:
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)
I'm looking for an elegant (pythonic) method to remove these redundant lines from having to be rewritten each time:
start = time.time()
- Should just automatically capture the start time at function launch.time.time() - start
Should use previously captured start time and infer current time from now()
. (Ideally elapsed_str()
would be callable with zero arguments.)My specific use case is to generate large datasets in the data science / data engineering field. Runtimes could be anywhere from seconds to days, and it is critical that (1) logs are easily searchable (for the word "elapsed" in this case) and (2) that the developer cost of adding the logs is very low (since we don't know ahead of time which jobs may be slow and we may not be able to modify source code once we identify a performance problem).
Upvotes: 3
Views: 2954
Reputation: 2571
This may be overkill for others' use cases but the solution I found required a few difficult hurtles and I'll document them here for anyone who wants to accomplish something similar.
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
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
Sample usage:
@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)
The main advantages versus simpler decorator solutions are:
desc_detail
, in which case the function would get evaluated both before and after function execution. This could eventually be extended to use a callback functions to count rows in created data table (for instance) and to include the table row counts in the completion log.Upvotes: 3
Reputation: 911
The recommended way is to use time.perf_counter()
and time.perf_counter_ns()
since 3.7.
In order to measure runtime of functions it is comfortable to use a decorator. For example the following one:
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) The code of this decorator is slightly modified from sosw
package
Upvotes: 3
Reputation: 46
If I understood you correctly you could write a decorator that will time the function.
A good example here: https://stackoverflow.com/a/5478448/6001492
Upvotes: 2