diff --git a/src/common/tools/context_queries/Topology.py b/src/common/tools/context_queries/Topology.py index 7759916b5aac78a18f12bf24251abbeafff16ecc..caf03ed0eb5271aa6e00a2c107a06f9e496d37dc 100644 --- a/src/common/tools/context_queries/Topology.py +++ b/src/common/tools/context_queries/Topology.py @@ -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 @@ -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