def __call__()

in idb/common/logging.py [0:0]

• 70 lines of code
• 8 McCabe index (conditional complexity)

    def __call__(self, function) -> Any:  # pyre-ignore
        _name = self.name or function.__name__

        @functools.wraps(function)
        async def _async_wrapper(*args: Any, **kwargs: Any) -> Any:  # pyre-ignore
            logger.debug(f"{_name} called")
            (_metadata, start) = _initial_info(args, self.metadata)
            await plugin.before_invocation(name=_name, metadata=_metadata)
            try:
                value = await function(*args, **kwargs)
                logger.debug(f"{_name} succeeded")
                await plugin.after_invocation(
                    name=_name,
                    duration=int((time.time() - start) * 1000),
                    metadata=_metadata,
                )
                return value
            except CancelledError as ex:
                logger.debug(f"{_name} cancelled")
                _metadata["cancelled"] = True
                await plugin.after_invocation(
                    name=_name,
                    duration=int((time.time() - start) * 1000),
                    metadata=_metadata,
                )
                raise ex
            except Exception as ex:
                logger.debug(f"{_name} failed")
                await plugin.failed_invocation(
                    name=_name,
                    duration=int((time.time() - start) * 1000),
                    exception=ex,
                    metadata=_metadata,
                )
                raise ex

        @functools.wraps(function)
        async def _async_gen_wrapper(*args, **kwargs) -> Any:  # pyre-ignore
            logger.debug(f"{_name} started")
            (_metadata, start) = _initial_info(args, self.metadata)
            await plugin.before_invocation(name=_name, metadata=_metadata)
            try:
                async for value in function(*args, **kwargs):
                    yield value
                logger.debug(f"{_name} finished")
                await plugin.after_invocation(
                    name=_name,
                    duration=int((time.time() - start) * 1000),
                    metadata=_metadata,
                )
            except CancelledError as ex:
                logger.debug(f"{_name} cancelled")
                _metadata["cancelled"] = True
                await plugin.after_invocation(
                    name=_name,
                    duration=int((time.time() - start) * 1000),
                    metadata=_metadata,
                )
                raise ex
            except Exception as ex:
                logger.debug(f"{_name} failed")
                await plugin.failed_invocation(
                    name=_name,
                    duration=int((time.time() - start) * 1000),
                    exception=ex,
                    metadata=_metadata,
                )
                raise ex

        if inspect.isasyncgenfunction(function):
            return _async_gen_wrapper
        else:
            return _async_wrapper