+ @staticmethod
+ def _create_task(
+ deployment_info: Dict[str, Any],
+ target_id: str,
+ item: str,
+ action: str,
+ target_record: str,
+ target_record_id: str,
+ extra_dict: Dict[str, Any] = None,
+ ) -> Dict[str, Any]:
+ """Function to create task dict from deployment information.
+
+ Args:
+ deployment_info (Dict[str, Any]): [description]
+ target_id (str): [description]
+ item (str): [description]
+ action (str): [description]
+ target_record (str): [description]
+ target_record_id (str): [description]
+ extra_dict (Dict[str, Any], optional): [description]. Defaults to None.
+
+ Returns:
+ Dict[str, Any]: [description]
+ """
+ task = {
+ "target_id": target_id, # it will be removed before pushing at database
+ "action_id": deployment_info.get("action_id"),
+ "nsr_id": deployment_info.get("nsr_id"),
+ "task_id": f"{deployment_info.get('action_id')}:{deployment_info.get('task_index')}",
+ "status": "SCHEDULED",
+ "action": action,
+ "item": item,
+ "target_record": target_record,
+ "target_record_id": target_record_id,
+ }
+
+ if extra_dict:
+ task.update(extra_dict) # params, find_params, depends_on
+
+ deployment_info["task_index"] = deployment_info.get("task_index", 0) + 1
+
+ return task
+