From bb094bf1bca8853fa691e439e14af535ddc7fdd1 Mon Sep 17 00:00:00 2001
From: gifrerenom <lluis.gifre@cttc.es>
Date: Thu, 16 Feb 2023 17:15:15 +0000
Subject: [PATCH] Common - Context Queries:
- Added method to get Context
- Added method to get Link
---
src/common/tools/context_queries/Context.py | 18 +++++++++++++++++-
src/common/tools/context_queries/Link.py | 19 +++++++++++++++++--
2 files changed, 34 insertions(+), 3 deletions(-)
diff --git a/src/common/tools/context_queries/Context.py b/src/common/tools/context_queries/Context.py
index d28ca3991..a627b9ba5 100644
--- a/src/common/tools/context_queries/Context.py
+++ b/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
diff --git a/src/common/tools/context_queries/Link.py b/src/common/tools/context_queries/Link.py
index 83a878bde..291cdcf37 100644
--- a/src/common/tools/context_queries/Link.py
+++ b/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}
--
GitLab