def duration_time()

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