def wrap_create_channel(create_channel_func, tracer=None):
    """Wrap the google.api_core.grpc_helpers.create_channel."""
    def call(*args, **kwargs):
        channel = create_channel_func(*args, **kwargs)

        try:
            target = kwargs.get('target')
            tracer_interceptor = OpenCensusClientInterceptor(tracer, target)
            intercepted_channel = grpc.intercept_channel(
                channel, tracer_interceptor)
            return intercepted_channel  # pragma: NO COVER
        except Exception:
            log.warning(
                'Failed to wrap create_channel, '
                'clientlibs grpc calls not traced.')
            return channel
    return call
def wrap_create_channel ( create_channel_func , tracer = None ) : def call ( * args , ** kwargs ) : channel = create_channel_func ( * args , ** kwargs ) try : target = kwargs . get ( 'target' ) tracer_interceptor = OpenCensusClientInterceptor ( tracer , target ) intercepted_channel = grpc . intercept_channel ( channel , tracer_interceptor ) return intercepted_channel except Exception : log . warning ( 'Failed to wrap create_channel, ' 'clientlibs grpc calls not traced.' ) return channel return call
Wrap the google.api_core.grpc_helpers.create_channel.