+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}"
+ )
+
+
projects / osm / NBI.git / blobdiff