Common - Context Queries:

- Added method to get Context
- Added method to get Link

src/common/tools/context_queries/Context.py

@@ -12,7 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from common.proto.context_pb2 import Context, Empty
+import grpc
+from typing import Optional
+from common.proto.context_pb2 import Context, ContextId, Empty
 from common.tools.object_factory.Context import json_context
 from context.client.ContextClient import ContextClient
 
@@ -23,3 +25,17 @@ def create_context(
     existing_context_uuids = {context_id.context_uuid.uuid for context_id in existing_context_ids.context_ids}
     if context_uuid in existing_context_uuids: return
     context_client.SetContext(Context(**json_context(context_uuid)))
+
+def get_context(context_client : ContextClient, context_uuid : str, rw_copy : bool = False) -> Optional[Context]:
+    try:
+        # pylint: disable=no-member
+        context_id = ContextId()
+        context_id.context_uuid.uuid = context_uuid
+        ro_context = context_client.GetContext(context_id)
+        if not rw_copy: return ro_context
+        rw_context = Context()
+        rw_context.CopyFrom(ro_context)
+        return rw_context
+    except grpc.RpcError:
+        #LOGGER.exception('Unable to get Context({:s})'.format(str(context_uuid)))
+        return None

src/common/tools/context_queries/Link.py

@@ -12,11 +12,26 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import List, Set
-from common.proto.context_pb2 import ContextId, Empty, Link, Topology, TopologyId
+import grpc
+from typing import List, Optional, Set
+from common.proto.context_pb2 import ContextId, Empty, Link, LinkId, Topology, TopologyId
 from common.tools.object_factory.Topology import json_topology_id
 from context.client.ContextClient import ContextClient
 
+def get_link(context_client : ContextClient, link_uuid : str, rw_copy : bool = False) -> Optional[Link]:
+    try:
+        # pylint: disable=no-member
+        link_id = LinkId()
+        link_id.link_uuid.uuid = link_uuid
+        ro_link = context_client.GetLink(link_id)
+        if not rw_copy: return ro_link
+        rw_link = Link()
+        rw_link.CopyFrom(ro_link)
+        return rw_link
+    except grpc.RpcError:
+        #LOGGER.exception('Unable to get Link({:s})'.format(str(link_uuid)))
+        return None
+
 def get_existing_link_uuids(context_client : ContextClient) -> Set[str]:
     existing_link_ids = context_client.ListLinkIds(Empty())
     existing_link_uuids = {link_id.link_uuid.uuid for link_id in existing_link_ids.link_ids}