X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_nbi%2Fbase_topic.py;h=820bd8e961c82883bdca871efddc35b7e5542e01;hb=7802ff80245ba7ba6055bc927b91e4f8b1f42542;hp=722ff594c253aaf90c4fcaa2835a30f4a75b92f1;hpb=4568a372eb5a204e04d917213de03ec51f9110c1;p=osm%2FNBI.git

diff --git a/osm_nbi/base_topic.py b/osm_nbi/base_topic.py
index 722ff59..820bd8e 100644
--- a/osm_nbi/base_topic.py
+++ b/osm_nbi/base_topic.py
@@ -17,7 +17,7 @@ import logging
 from uuid import uuid4
 from http import HTTPStatus
 from time import time
-from osm_common.dbbase import deep_update_rfc7396
+from osm_common.dbbase import deep_update_rfc7396, DbException
 from osm_nbi.validation import validate_input, ValidationError, is_valid_uuid
 from yaml import safe_load, YAMLError
 
@@ -30,6 +30,20 @@ class EngineException(Exception):
         super(Exception, self).__init__(message)
 
 
+class NBIBadArgumentsException(Exception):
+    """
+    Bad argument values exception
+    """
+
+    def __init__(self, message: str = "", bad_args: list = None):
+        Exception.__init__(self, message)
+        self.message = message
+        self.bad_args = bad_args
+
+    def __str__(self):
+        return "{}, Bad arguments: {}".format(self.message, self.bad_args)
+
+
 def deep_get(target_dict, key_list):
     """
     Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
@@ -45,6 +59,85 @@ def deep_get(target_dict, key_list):
     return target_dict
 
 
+def detect_descriptor_usage(descriptor: dict, db_collection: str, db: object) -> bool:
+    """Detect the descriptor usage state.
+
+    Args:
+        descriptor (dict):   VNF or NS Descriptor as dictionary
+        db_collection (str):   collection name which is looked for in DB
+        db (object):   name of db object
+
+    Returns:
+        True if descriptor is in use else None
+
+    """
+    try:
+        if not descriptor:
+            raise NBIBadArgumentsException(
+                "Argument is mandatory and can not be empty", "descriptor"
+            )
+
+        if not db:
+            raise NBIBadArgumentsException("A valid DB object should be provided", "db")
+
+        search_dict = {
+            "vnfds": ("vnfrs", "vnfd-id"),
+            "nsds": ("nsrs", "nsd-id"),
+        }
+
+        if db_collection not in search_dict:
+            raise NBIBadArgumentsException(
+                "db_collection should be equal to vnfds or nsds", "db_collection"
+            )
+
+        record_list = db.get_list(
+            search_dict[db_collection][0],
+            {search_dict[db_collection][1]: descriptor["_id"]},
+        )
+
+        if record_list:
+            return True
+
+    except (DbException, KeyError, NBIBadArgumentsException) as error:
+        raise EngineException(
+            f"Error occured while detecting the descriptor usage: {error}"
+        )
+
+
+def update_descriptor_usage_state(
+    descriptor: dict, db_collection: str, db: object
+) -> None:
+    """Updates the descriptor usage state.
+
+    Args:
+        descriptor (dict):   VNF or NS Descriptor as dictionary
+        db_collection (str):   collection name which is looked for in DB
+        db (object):   name of db object
+
+    Returns:
+        None
+
+    """
+    try:
+        descriptor_update = {
+            "_admin.usageState": "NOT_IN_USE",
+        }
+
+        if detect_descriptor_usage(descriptor, db_collection, db):
+            descriptor_update = {
+                "_admin.usageState": "IN_USE",
+            }
+
+        db.set_one(
+            db_collection, {"_id": descriptor["_id"]}, update_dict=descriptor_update
+        )
+
+    except (DbException, KeyError, NBIBadArgumentsException) as error:
+        raise EngineException(
+            f"Error occured while updating the descriptor usage state: {error}"
+        )
+
+
 def get_iterable(input_var):
     """
     Returns an iterable, in case input_var is None it just returns an empty tuple
@@ -411,11 +504,12 @@ class BaseTopic:
         # Projection was moved to child classes
         return data
 
-    def show(self, session, _id, api_req=False):
+    def show(self, session, _id, filter_q=None, api_req=False):
         """
         Get complete information on an topic
         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
         :param _id: server internal id
+        :param filter_q: dict: query parameter
         :param api_req: True if this call is serving an external API request. False if serving internal request.
         :return: dictionary, raise exception if not found.
         """