from typing import Iterator import grpc, logging from common.tools.client.RetryDecorator import retry, delay_exponential from context.proto.context_pb2 import \ Context, ContextEvent, ContextId, ContextIdList, ContextList, \ Device, DeviceEvent, DeviceId, DeviceIdList, DeviceList, \ Empty, \ Link, LinkEvent, LinkId, LinkIdList, LinkList, \ Service, ServiceEvent, ServiceId, ServiceIdList, ServiceList, \ Topology, TopologyEvent, TopologyId, TopologyIdList, TopologyList from context.proto.context_pb2_grpc import ContextServiceStub LOGGER = logging.getLogger(__name__) MAX_RETRIES = 15 DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0) class ContextClient: def __init__(self, address, port): self.endpoint = '{:s}:{:s}'.format(str(address), str(port)) LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint)) self.channel = None self.stub = None self.connect() LOGGER.debug('Channel created') def connect(self): self.channel = grpc.insecure_channel(self.endpoint) self.stub = ContextServiceStub(self.channel) def close(self): if self.channel is not None: self.channel.close() self.channel = None self.stub = None @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListContextIds(self, request: Empty) -> ContextIdList: LOGGER.debug('ListContextIds request: {:s}'.format(str(request))) response = self.stub.ListContextIds(request) LOGGER.debug('ListContextIds result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListContexts(self, request: Empty) -> ContextList: LOGGER.debug('ListContexts request: {:s}'.format(str(request))) response = self.stub.ListContexts(request) LOGGER.debug('ListContexts result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetContext(self, request: ContextId) -> Context: LOGGER.debug('GetContext request: {:s}'.format(str(request))) response = self.stub.GetContext(request) LOGGER.debug('GetContext result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def SetContext(self, request: Context) -> ContextId: LOGGER.debug('SetContext request: {:s}'.format(str(request))) response = self.stub.SetContext(request) LOGGER.debug('SetContext result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def RemoveContext(self, request: ContextId) -> Empty: LOGGER.debug('RemoveContext request: {:s}'.format(str(request))) response = self.stub.RemoveContext(request) LOGGER.debug('RemoveContext result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetContextEvents(self, request: Empty) -> Iterator[ContextEvent]: LOGGER.debug('GetContextEvents request: {:s}'.format(str(request))) response = self.stub.GetContextEvents(request) LOGGER.debug('GetContextEvents result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListTopologyIds(self, request: ContextId) -> TopologyIdList: LOGGER.debug('ListTopologyIds request: {:s}'.format(str(request))) response = self.stub.ListTopologyIds(request) LOGGER.debug('ListTopologyIds result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListTopologies(self, request: ContextId) -> TopologyList: LOGGER.debug('ListTopologies request: {:s}'.format(str(request))) response = self.stub.ListTopologies(request) LOGGER.debug('ListTopologies result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetTopology(self, request: TopologyId) -> Topology: LOGGER.debug('GetTopology request: {:s}'.format(str(request))) response = self.stub.GetTopology(request) LOGGER.debug('GetTopology result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def SetTopology(self, request: Topology) -> TopologyId: LOGGER.debug('SetTopology request: {:s}'.format(str(request))) response = self.stub.SetTopology(request) LOGGER.debug('SetTopology result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def RemoveTopology(self, request: TopologyId) -> Empty: LOGGER.debug('RemoveTopology request: {:s}'.format(str(request))) response = self.stub.RemoveTopology(request) LOGGER.debug('RemoveTopology result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetTopologyEvents(self, request: Empty) -> Iterator[TopologyEvent]: LOGGER.debug('GetTopologyEvents request: {:s}'.format(str(request))) response = self.stub.GetTopologyEvents(request) LOGGER.debug('GetTopologyEvents result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListDeviceIds(self, request: Empty) -> DeviceIdList: LOGGER.debug('ListDeviceIds request: {:s}'.format(str(request))) response = self.stub.ListDeviceIds(request) LOGGER.debug('ListDeviceIds result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListDevices(self, request: Empty) -> DeviceList: LOGGER.debug('ListDevices request: {:s}'.format(str(request))) response = self.stub.ListDevices(request) LOGGER.debug('ListDevices result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetDevice(self, request: DeviceId) -> Device: LOGGER.debug('GetDevice request: {:s}'.format(str(request))) response = self.stub.GetDevice(request) LOGGER.debug('GetDevice result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def SetDevice(self, request: Device) -> DeviceId: LOGGER.debug('SetDevice request: {:s}'.format(str(request))) response = self.stub.SetDevice(request) LOGGER.debug('SetDevice result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def RemoveDevice(self, request: DeviceId) -> Empty: LOGGER.debug('RemoveDevice request: {:s}'.format(str(request))) response = self.stub.RemoveDevice(request) LOGGER.debug('RemoveDevice result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetDeviceEvents(self, request: Empty) -> Iterator[DeviceEvent]: LOGGER.debug('GetDeviceEvents request: {:s}'.format(str(request))) response = self.stub.GetDeviceEvents(request) LOGGER.debug('GetDeviceEvents result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListLinkIds(self, request: Empty) -> LinkIdList: LOGGER.debug('ListLinkIds request: {:s}'.format(str(request))) response = self.stub.ListLinkIds(request) LOGGER.debug('ListLinkIds result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListLinks(self, request: Empty) -> LinkList: LOGGER.debug('ListLinks request: {:s}'.format(str(request))) response = self.stub.ListLinks(request) LOGGER.debug('ListLinks result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetLink(self, request: LinkId) -> Link: LOGGER.debug('GetLink request: {:s}'.format(str(request))) response = self.stub.GetLink(request) LOGGER.debug('GetLink result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def SetLink(self, request: Link) -> LinkId: LOGGER.debug('SetLink request: {:s}'.format(str(request))) response = self.stub.SetLink(request) LOGGER.debug('SetLink result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def RemoveLink(self, request: LinkId) -> Empty: LOGGER.debug('RemoveLink request: {:s}'.format(str(request))) response = self.stub.RemoveLink(request) LOGGER.debug('RemoveLink result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetLinkEvents(self, request: Empty) -> Iterator[LinkEvent]: LOGGER.debug('GetLinkEvents request: {:s}'.format(str(request))) response = self.stub.GetLinkEvents(request) LOGGER.debug('GetLinkEvents result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListServiceIds(self, request: ContextId) -> ServiceIdList: LOGGER.debug('ListServiceIds request: {:s}'.format(str(request))) response = self.stub.ListServiceIds(request) LOGGER.debug('ListServiceIds result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def ListServices(self, request: ContextId) -> ServiceList: LOGGER.debug('ListServices request: {:s}'.format(str(request))) response = self.stub.ListServices(request) LOGGER.debug('ListServices result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetService(self, request: ServiceId) -> Service: LOGGER.debug('GetService request: {:s}'.format(str(request))) response = self.stub.GetService(request) LOGGER.debug('GetService result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def SetService(self, request: Service) -> ServiceId: LOGGER.debug('SetService request: {:s}'.format(str(request))) response = self.stub.SetService(request) LOGGER.debug('SetService result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def RemoveService(self, request: ServiceId) -> Empty: LOGGER.debug('RemoveService request: {:s}'.format(str(request))) response = self.stub.RemoveService(request) LOGGER.debug('RemoveService result: {:s}'.format(str(response))) return response @retry(exceptions=set(), max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect') def GetServiceEvents(self, request: Empty) -> Iterator[ServiceEvent]: LOGGER.debug('GetServiceEvents request: {:s}'.format(str(request))) response = self.stub.GetServiceEvents(request) LOGGER.debug('GetServiceEvents result: {:s}'.format(str(response))) return response