Loading src/common/tools/context_queries/Topology.py +19 −1 Original line number Diff line number Diff line Loading @@ -15,7 +15,7 @@ import grpc, logging from typing import List, Optional from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ContextId, Topology, TopologyId from common.proto.context_pb2 import ContextId, Topology, TopologyDetails, TopologyId from common.tools.object_factory.Context import json_context_id from common.tools.object_factory.Topology import json_topology from context.client.ContextClient import ContextClient Loading Loading @@ -61,3 +61,21 @@ def get_topology( except grpc.RpcError: #LOGGER.exception('Unable to get topology({:s} / {:s})'.format(str(context_uuid), str(topology_uuid))) return None def get_topology_details( context_client : ContextClient, topology_uuid : str, context_uuid : str = DEFAULT_CONTEXT_NAME, rw_copy : bool = False ) -> Optional[Topology]: try: # pylint: disable=no-member topology_id = TopologyId() topology_id.context_id.context_uuid.uuid = context_uuid topology_id.topology_uuid.uuid = topology_uuid ro_topology_details = context_client.GetTopologyDetails(topology_id) if not rw_copy: return ro_topology_details rw_topology_details = TopologyDetails() rw_topology_details.CopyFrom(ro_topology_details) return rw_topology_details except grpc.RpcError: #LOGGER.exception('Unable to get topology({:s} / {:s})'.format(str(context_uuid), str(topology_uuid))) return None Loading
src/common/tools/context_queries/Topology.py +19 −1 Original line number Diff line number Diff line Loading @@ -15,7 +15,7 @@ import grpc, logging from typing import List, Optional from common.Constants import DEFAULT_CONTEXT_NAME from common.proto.context_pb2 import ContextId, Topology, TopologyId from common.proto.context_pb2 import ContextId, Topology, TopologyDetails, TopologyId from common.tools.object_factory.Context import json_context_id from common.tools.object_factory.Topology import json_topology from context.client.ContextClient import ContextClient Loading Loading @@ -61,3 +61,21 @@ def get_topology( except grpc.RpcError: #LOGGER.exception('Unable to get topology({:s} / {:s})'.format(str(context_uuid), str(topology_uuid))) return None def get_topology_details( context_client : ContextClient, topology_uuid : str, context_uuid : str = DEFAULT_CONTEXT_NAME, rw_copy : bool = False ) -> Optional[Topology]: try: # pylint: disable=no-member topology_id = TopologyId() topology_id.context_id.context_uuid.uuid = context_uuid topology_id.topology_uuid.uuid = topology_uuid ro_topology_details = context_client.GetTopologyDetails(topology_id) if not rw_copy: return ro_topology_details rw_topology_details = TopologyDetails() rw_topology_details.CopyFrom(ro_topology_details) return rw_topology_details except grpc.RpcError: #LOGGER.exception('Unable to get topology({:s} / {:s})'.format(str(context_uuid), str(topology_uuid))) return None