Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, channel: grpclib.client.Channel) -> None:
self.UnaryUnary = grpclib.client.UnaryUnaryMethod(
channel,
'/dummy.DummyService/UnaryUnary',
dummy_pb2.DummyRequest,
dummy_pb2.DummyReply,
)
self.UnaryStream = grpclib.client.UnaryStreamMethod(
channel,
'/dummy.DummyService/UnaryStream',
dummy_pb2.DummyRequest,
dummy_pb2.DummyReply,
)
self.StreamUnary = grpclib.client.StreamUnaryMethod(
channel,
'/dummy.DummyService/StreamUnary',
dummy_pb2.DummyRequest,
dummy_pb2.DummyReply,
)
self.StreamStream = grpclib.client.StreamStreamMethod(
channel,
'/dummy.DummyService/StreamStream',
dummy_pb2.DummyRequest,
dummy_pb2.DummyReply,
def __init__(self, channel: grpclib.client.Channel) -> None:
self.Plaster = grpclib.client.UnaryUnaryMethod(
channel,
'/Bombed/Plaster',
bombed_pb2.SavoysRequest,
bombed_pb2.SavoysReply,
)
self.Benzine = grpclib.client.UnaryStreamMethod(
channel,
'/Bombed/Benzine',
bombed_pb2.SavoysRequest,
bombed_pb2.GoowyChunk,
)
self.Anginal = grpclib.client.StreamUnaryMethod(
channel,
'/Bombed/Anginal',
bombed_pb2.UnyoungChunk,
bombed_pb2.SavoysReply,
)
self.Devilry = grpclib.client.StreamStreamMethod(
channel,
'/Bombed/Devilry',
bombed_pb2.UnyoungChunk,
bombed_pb2.GoowyChunk,
def __init__(self, channel: grpclib.client.Channel) -> None:
self.Check = grpclib.client.UnaryUnaryMethod(
channel,
'/grpc.health.v1.Health/Check',
grpclib.health.v1.health_pb2.HealthCheckRequest,
grpclib.health.v1.health_pb2.HealthCheckResponse,
)
self.Watch = grpclib.client.UnaryStreamMethod(
channel,
'/grpc.health.v1.Health/Watch',
grpclib.health.v1.health_pb2.HealthCheckRequest,
grpclib.health.v1.health_pb2.HealthCheckResponse,
)
def __init__(self, channel: grpclib.client.Channel) -> None:
self.UnaryUnaryGreeting = grpclib.client.UnaryUnaryMethod(
channel,
'/helloworld.Greeter/UnaryUnaryGreeting',
helloworld.helloworld_pb2.HelloRequest,
helloworld.helloworld_pb2.HelloReply,
)
self.UnaryStreamGreeting = grpclib.client.UnaryStreamMethod(
channel,
'/helloworld.Greeter/UnaryStreamGreeting',
helloworld.helloworld_pb2.HelloRequest,
helloworld.helloworld_pb2.HelloReply,
)
self.StreamUnaryGreeting = grpclib.client.StreamUnaryMethod(
channel,
'/helloworld.Greeter/StreamUnaryGreeting',
helloworld.helloworld_pb2.HelloRequest,
helloworld.helloworld_pb2.HelloReply,
)
self.StreamStreamGreeting = grpclib.client.StreamStreamMethod(
channel,
'/helloworld.Greeter/StreamStreamGreeting',
helloworld.helloworld_pb2.HelloRequest,
helloworld.helloworld_pb2.HelloReply,
def __init__(self, channel: grpclib.client.Channel) -> None:
self.UnaryUnaryGreeting = grpclib.client.UnaryUnaryMethod(
channel,
'/helloworld.Greeter/UnaryUnaryGreeting',
streaming.helloworld_pb2.HelloRequest,
streaming.helloworld_pb2.HelloReply,
)
self.UnaryStreamGreeting = grpclib.client.UnaryStreamMethod(
channel,
'/helloworld.Greeter/UnaryStreamGreeting',
streaming.helloworld_pb2.HelloRequest,
streaming.helloworld_pb2.HelloReply,
)
self.StreamUnaryGreeting = grpclib.client.StreamUnaryMethod(
channel,
'/helloworld.Greeter/StreamUnaryGreeting',
streaming.helloworld_pb2.HelloRequest,
streaming.helloworld_pb2.HelloReply,
)
self.StreamStreamGreeting = grpclib.client.StreamStreamMethod(
channel,
'/helloworld.Greeter/StreamStreamGreeting',
streaming.helloworld_pb2.HelloRequest,
streaming.helloworld_pb2.HelloReply,
buf.add('')
buf.add('')
buf.add('class {}Stub:', service.name)
with buf.indent():
buf.add('')
buf.add('def __init__(self, channel: {}.{}) -> None:'
.format(client.__name__, client.Channel.__name__))
with buf.indent():
for method in service.methods:
name, cardinality, request_type, reply_type = method
full_name = '/{}/{}'.format(service_name, name)
method_cls: type
if cardinality is const.Cardinality.UNARY_UNARY:
method_cls = client.UnaryUnaryMethod
elif cardinality is const.Cardinality.UNARY_STREAM:
method_cls = client.UnaryStreamMethod
elif cardinality is const.Cardinality.STREAM_UNARY:
method_cls = client.StreamUnaryMethod
elif cardinality is const.Cardinality.STREAM_STREAM:
method_cls = client.StreamStreamMethod
else:
raise TypeError(cardinality)
method_cls = cast(type, method_cls) # FIXME: redundant
buf.add('self.{} = {}.{}('.format(name, client.__name__,
method_cls.__name__))
with buf.indent():
buf.add('channel,')
buf.add('{!r},'.format(full_name))
buf.add('{},', request_type)
buf.add('{},', reply_type)
buf.add(')')
return buf.content()