# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
from nvflare.private.fed.protos import admin_pb2 as nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2
[docs]class AdminCommunicatingStub(object):
"""import "google/protobuf/timestamp.proto";
import "google/protobuf/struct.proto";
The admin communication interfaces
"""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.Retrieve = channel.unary_unary(
'/admin.AdminCommunicating/Retrieve',
request_serializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Client.SerializeToString,
response_deserializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Messages.FromString,
)
self.SendReply = channel.unary_unary(
'/admin.AdminCommunicating/SendReply',
request_serializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Reply.SerializeToString,
response_deserializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Empty.FromString,
)
self.SendResult = channel.unary_unary(
'/admin.AdminCommunicating/SendResult',
request_serializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Reply.SerializeToString,
response_deserializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Empty.FromString,
)
[docs]class AdminCommunicatingServicer(object):
"""import "google/protobuf/timestamp.proto";
import "google/protobuf/struct.proto";
The admin communication interfaces
"""
[docs] def Retrieve(self, request, context):
"""client retrieve requests.
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
[docs] def SendReply(self, request, context):
"""client send reply to server
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
[docs] def SendResult(self, request, context):
"""client send process results to server
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
[docs]def add_AdminCommunicatingServicer_to_server(servicer, server):
rpc_method_handlers = {
'Retrieve': grpc.unary_unary_rpc_method_handler(
servicer.Retrieve,
request_deserializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Client.FromString,
response_serializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Messages.SerializeToString,
),
'SendReply': grpc.unary_unary_rpc_method_handler(
servicer.SendReply,
request_deserializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Reply.FromString,
response_serializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Empty.SerializeToString,
),
'SendResult': grpc.unary_unary_rpc_method_handler(
servicer.SendResult,
request_deserializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Reply.FromString,
response_serializer=nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Empty.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'admin.AdminCommunicating', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
# This class is part of an EXPERIMENTAL API.
[docs]class AdminCommunicating(object):
"""import "google/protobuf/timestamp.proto";
import "google/protobuf/struct.proto";
The admin communication interfaces
"""
[docs] @staticmethod
def Retrieve(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/admin.AdminCommunicating/Retrieve',
nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Client.SerializeToString,
nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Messages.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
[docs] @staticmethod
def SendReply(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/admin.AdminCommunicating/SendReply',
nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Reply.SerializeToString,
nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Empty.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)
[docs] @staticmethod
def SendResult(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(request, target, '/admin.AdminCommunicating/SendResult',
nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Reply.SerializeToString,
nvflare_dot_private_dot_fed_dot_protos_dot_admin__pb2.Empty.FromString,
options, channel_credentials,
insecure, call_credentials, compression, wait_for_ready, timeout, metadata)