in fbpcp/decorator/metrics.py [0:0]
def duration_time(metrics_name: str) -> Callable:
def wrap(f: Callable):
@functools.wraps(f)
def wrapper_sync(self: MetricsGetter, *args, **kwargs):
start = time.perf_counter_ns()
res = f(self, *args, **kwargs)
end = time.perf_counter_ns()
if self.has_metrics():
self.get_metrics().gauge(metrics_name, int((end - start) / 1e6))
return res
@functools.wraps(f)
async def wrapper_async(self: MetricsGetter, *args, **kwargs):
start = time.perf_counter_ns()
res = await f(self, *args, **kwargs)
end = time.perf_counter_ns()
if self.has_metrics():
self.get_metrics().gauge(metrics_name, int((end - start) / 1e6))
return res
return wrapper_async if asyncio.iscoroutinefunction(f) else wrapper_sync
return wrap