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
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
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
# 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.
"""