return task
+ @staticmethod
+ def _create_ro_task(
+ target_id: str,
+ task: Dict[str, Any],
+ ) -> Dict[str, Any]:
+ """Function to create an RO task from task information.
+
+ Args:
+ target_id (str): [description]
+ task (Dict[str, Any]): [description]
+
+ Returns:
+ Dict[str, Any]: [description]
+ """
+ now = time()
+
+ _id = task.get("task_id")
+ db_ro_task = {
+ "_id": _id,
+ "locked_by": None,
+ "locked_at": 0.0,
+ "target_id": target_id,
+ "vim_info": {
+ "created": False,
+ "created_items": None,
+ "vim_id": None,
+ "vim_name": None,
+ "vim_status": None,
+ "vim_details": None,
+ "refresh_at": None,
+ },
+ "modified_at": now,
+ "created_at": now,
+ "to_check_at": now,
+ "tasks": [task],
+ }
+
+ return db_ro_task
+
+ @staticmethod
+ def _process_image_params(
+ target_image: Dict[str, Any],
+ vim_info: Dict[str, Any],
+ target_record_id: str,
+ ) -> Dict[str, Any]:
+ """Function to process VDU image parameters.
+
+ Args:
+ target_image (Dict[str, Any]): [description]
+ vim_info (Dict[str, Any]): [description]
+ target_record_id (str): [description]
+
+ Returns:
+ Dict[str, Any]: [description]
+ """
+ find_params = {}
+
+ if target_image.get("image"):
+ find_params["filter_dict"] = {"name": target_image.get("image")}
+
+ if target_image.get("vim_image_id"):
+ find_params["filter_dict"] = {"id": target_image.get("vim_image_id")}
+
+ if target_image.get("image_checksum"):
+ find_params["filter_dict"] = {
+ "checksum": target_image.get("image_checksum")
+ }
+
+ return {"find_params": find_params}
+
def deploy(self, session, indata, version, nsr_id, *args, **kwargs):
self.logger.debug("ns.deploy nsr_id={} indata={}".format(nsr_id, indata))
validate_input(indata, deploy_schema)
index += 1
- def _create_ro_task(target_id, task):
- nonlocal action_id
- nonlocal task_index
- nonlocal now
-
- _id = task["task_id"]
- db_ro_task = {
- "_id": _id,
- "locked_by": None,
- "locked_at": 0.0,
- "target_id": target_id,
- "vim_info": {
- "created": False,
- "created_items": None,
- "vim_id": None,
- "vim_name": None,
- "vim_status": None,
- "vim_details": None,
- "refresh_at": None,
- },
- "modified_at": now,
- "created_at": now,
- "to_check_at": now,
- "tasks": [task],
- }
-
- return db_ro_task
-
- def _process_image_params(target_image, vim_info, target_record_id):
- find_params = {}
-
- if target_image.get("image"):
- find_params["filter_dict"] = {"name": target_image.get("image")}
-
- if target_image.get("vim_image_id"):
- find_params["filter_dict"] = {
- "id": target_image.get("vim_image_id")
- }
-
- if target_image.get("image_checksum"):
- find_params["filter_dict"] = {
- "checksum": target_image.get("image_checksum")
- }
-
- return {"find_params": find_params}
-
def _process_flavor_params(target_flavor, vim_info, target_record_id):
def _get_resource_allocation_params(quota_descriptor):
"""
db_update=db_nsr_update,
db_path="image",
item="image",
- process_params=_process_image_params,
+ process_params=Ns._process_image_params,
)
step = "process NS flavors"
):
# Create a ro_task
step = "Updating database, Creating ro_tasks"
- db_ro_task = _create_ro_task(target_id, db_task)
+ db_ro_task = Ns._create_ro_task(target_id, db_task)
nb_ro_tasks += 1
self.db.create("ro_tasks", db_ro_task)