def log_and_handle_exceptions()

in idb/grpc/client.py [0:0]

• 38 lines of code
• 11 McCabe index (conditional complexity)

def log_and_handle_exceptions(grpc_method_name: str):  # pyre-ignore

    is_native_swift_call = (
        os.environ.get("IDB_USE_SWIFT") == "YES"
        and grpc_method_name in extract_native_swift_methods_from_env()
    )
    metadata: LoggingMetadata = {
        "grpc_method_name": grpc_method_name,
        "grpc_native_swift_call": "yes" if is_native_swift_call else "no",
    }

    def decorating(func) -> Any:  # pyre-ignore:
        @functools.wraps(func)
        @log_call(name=func.__name__, metadata=metadata)
        async def func_wrapper(*args: Any, **kwargs: Any) -> Any:  # pyre-ignore
            try:
                return await func(*args, **kwargs)
            except GRPCError as e:
                raise IdbException(e.message) from e  # noqa B306
            except (ProtocolError, StreamTerminatedError) as e:
                raise IdbException(e.args) from e
            except OSError as e:
                raise IdbConnectionException(e.strerror)

        @functools.wraps(func)
        @log_call(name=func.__name__, metadata=metadata)
        # pyre-fixme[53]: Captured variable `func` is not annotated.
        # pyre-fixme[3]: Return annotation cannot be `Any`.
        async def func_wrapper_gen(*args: Any, **kwargs: Any) -> Any:
            try:
                async for item in func(*args, **kwargs):
                    yield item
            except GRPCError as e:
                raise IdbException(e.message) from e  # noqa B306
            except (ProtocolError, StreamTerminatedError) as e:
                raise IdbException(e.args) from e
            except OSError as e:
                raise IdbConnectionException(e.strerror)

        if inspect.isasyncgenfunction(func):
            return func_wrapper_gen
        else:
            return func_wrapper

    return decorating