added tasks objects

.changelog/4343.yml

@@ -0,0 +1,4 @@
+changes:
+- description: Modeled base_playbook tasks into objects to ease on the use in various flows.
+  type: internal
+pr_number: 4343

demisto_sdk/commands/common/constants.py

@@ -2178,6 +2178,17 @@ class IncidentState(StrEnum):
     ACKNOWLEDGED = "ACKNOWLEDGED"
 
 
+class PlaybookTaskType(StrEnum):
+    REGULAR = "regular"
+    PLAYBOOK = "playbook"
+    CONDITION = "condition"
+    START = "start"
+    TITLE = "title"
+    SECTION = "section"
+    STANDARD = "standard"
+    COLLECTION = "collection"
+
+
 # Used to format the writing of the yml/json file
 DEFAULT_JSON_INDENT = 4
 DEFAULT_YAML_INDENT = 0

demisto_sdk/commands/content_graph/objects/base_playbook.py

@@ -1,11 +1,14 @@
 from functools import cached_property
-from typing import Callable, Optional
+from typing import Callable, Dict, List, Optional
 
 import demisto_client
+from pydantic import BaseModel, Field
 
 from demisto_sdk.commands.common.constants import (
     MarketplaceVersions,
+    PlaybookTaskType,
 )
+from demisto_sdk.commands.common.tools import remove_nulls_from_dictionary, write_dict
 from demisto_sdk.commands.content_graph.common import ContentType
 from demisto_sdk.commands.content_graph.objects.content_item import ContentItem
 from demisto_sdk.commands.content_graph.parsers.related_files import (
@@ -17,8 +20,161 @@
 )
 
 
+class Task(BaseModel):
+    id: str
+    version: Optional[int] = None
+    name: Optional[str] = None
+    name_x2: Optional[str] = None
+    playbookName: Optional[str] = None
+    playbookName_x2: Optional[str] = None
+    playbookId: Optional[str] = None
+    playbookId_x2: Optional[str] = None
+    description: Optional[str] = None
+    description_x2: Optional[str] = None
+    scriptName: Optional[str] = None
+    scriptName_x2: Optional[str] = None
+    script: Optional[str] = None
+    script_x2: Optional[str] = None
+    tags: Optional[List[str]] = None
+    type: Optional[str] = None
+    iscommand: Optional[bool] = None
+    elasticcommonfields: Optional[Dict[str, str]] = None
+    brand: Optional[str] = None
+    issystemtask: Optional[bool] = None
+    clonedfrom: Optional[str] = None
+    name_xsoar: Optional[str] = None
+    name_marketplacev2: Optional[str] = None
+    name_xpanse: Optional[str] = None
+    name_xsoar_saas: Optional[str] = None
+    name_xsoar_on_prem: Optional[str] = None
+    description_xsoar: Optional[str] = None
+    description_marketplacev2: Optional[str] = None
+    description_xpanse: Optional[str] = None
+    description_xsoar_saas: Optional[str] = None
+    description_xsoar_on_prem: Optional[str] = None
+
+    @property
+    def to_raw_dict(self) -> Dict:
+        """Generate a dict representation of the Task object.
+
+        Returns:
+            Dict: The dict representation of the Task object.
+        """
+        task = {
+            "id": self.id,
+            "version": self.version,
+            "name": self.name,
+            "name_x2": self.name_x2,
+            "playbookName": self.playbookName,
+            "playbookName_x2": self.playbookName_x2,
+            "playbookId": self.playbookId,
+            "playbookId_x2": self.playbookId_x2,
+            "description": self.description,
+            "description_x2": self.description_x2,
+            "scriptName": self.scriptName,
+            "scriptName_x2": self.scriptName_x2,
+            "script": self.script,
+            "script_x2": self.script_x2,
+            "tags": self.tags,
+            "type": self.type,
+            "iscommand": self.iscommand,
+            "elasticcommonfields": self.elasticcommonfields,
+            "brand": self.brand,
+            "issystemtask": self.issystemtask,
+            "clonedfrom": self.clonedfrom,
+            "name_xsoar": self.name_xsoar,
+            "name_marketplacev2": self.name_marketplacev2,
+            "name_xpanse": self.name_xpanse,
+            "name_xsoar_saas": self.name_xsoar_saas,
+            "name_xsoar_on_prem": self.name_xsoar_on_prem,
+            "description_xsoar": self.description_xsoar,
+            "description_marketplacev2": self.description_marketplacev2,
+            "description_xpanse": self.description_xpanse,
+            "description_xsoar_saas": self.description_xsoar_saas,
+            "description_xsoar_on_prem": self.description_xsoar_on_prem,
+        }
+        remove_nulls_from_dictionary(task)
+        return task
+
+
+class TaskConfig(BaseModel):
+    id: str
+    taskid: str
+    type: Optional[PlaybookTaskType] = None
+    form: Optional[Dict] = None
+    message: Optional[Dict] = None
+    defaultassigneecomplex: Optional[Dict] = None
+    sla: Optional[Dict] = None
+    slareminder: Optional[Dict] = None
+    quietmode: Optional[int] = None
+    restrictedcompletion: Optional[bool] = None
+    scriptarguments: Optional[Dict] = None
+    timertriggers: Optional[List] = None
+    ignoreworker: Optional[bool] = None
+    skipunavailable: Optional[bool] = None
+    isoversize: Optional[bool] = None
+    isautoswitchedtoquietmode: Optional[bool] = None
+    quiet: Optional[bool] = None
+    evidencedata: Optional[dict] = None
+    task: Task
+    note: Optional[bool] = None
+    nexttasks: Optional[Dict[str, List[str]]] = None
+    loop: Optional[Dict] = None
+    conditions: Optional[List[dict]] = None
+    view: Optional[str] = None
+    results: Optional[List[str]] = None
+    continueonerror: Optional[bool] = None
+    continueonerrortype: Optional[str] = None
+    reputationcalc: Optional[int] = None
+    separatecontext: Optional[bool] = None
+    fieldMapping: Optional[List] = None
+
+    @property
+    def to_raw_dict(self) -> Dict:
+        """Generate a dict representation of the TaskConfig object.
+
+        Returns:
+            Dict: The dict representation of the TaskConfig object.
+        """
+        task_config = {
+            "id": self.id,
+            "taskid": self.taskid,
+            "type": self.type,
+            "form": self.form,
+            "message": self.message,
+            "defaultassigneecomplex": self.defaultassigneecomplex,
+            "sla": self.sla,
+            "slareminder": self.slareminder,
+            "quietmode": self.quietmode,
+            "restrictedcompletion": self.restrictedcompletion,
+            "scriptarguments": self.scriptarguments,
+            "timertriggers": self.timertriggers,
+            "ignoreworker": self.ignoreworker,
+            "skipunavailable": self.skipunavailable,
+            "isoversize": self.isoversize,
+            "isautoswitchedtoquietmode": self.isautoswitchedtoquietmode,
+            "quiet": self.quiet,
+            "evidencedata": self.evidencedata,
+            "task": self.task.to_raw_dict,
+            "note": self.note,
+            "nexttasks": self.nexttasks,
+            "loop": self.loop,
+            "conditions": self.conditions,
+            "view": self.view,
+            "results": self.results,
+            "continueonerror": self.continueonerror,
+            "continueonerrortype": self.continueonerrortype,
+            "reputationcalc": self.reputationcalc,
+            "separatecontext": self.separatecontext,
+            "fieldMapping": self.fieldMapping,
+        }
+        remove_nulls_from_dictionary(task_config)
+        return task_config
+
+
 class BasePlaybook(ContentItem, content_type=ContentType.PLAYBOOK):  # type: ignore[call-arg]
     version: Optional[int] = 0
+    tasks: Dict[str, TaskConfig] = Field([], exclude=True)
 
     def summary(
         self,
@@ -48,6 +204,15 @@ def prepare_for_upload(
     def _client_upload_method(cls, client: demisto_client) -> Callable:
         return client.import_playbook
 
+    def save(self):
+        super().save()
+        data = self.data
+        data["tasks"] = {
+            task_id: task_config.to_raw_dict
+            for task_id, task_config in self.tasks.items()
+        }
+        write_dict(self.path, data, indent=4)
+
     @cached_property
     def readme(self) -> ReadmeRelatedFile:
         return ReadmeRelatedFile(self.path, is_pack_readme=False, git_sha=self.git_sha)

demisto_sdk/commands/content_graph/objects/playbook.py

@@ -1,15 +1,12 @@
 from pathlib import Path
 
-from pydantic import Field
-
 from demisto_sdk.commands.common.constants import TEST_PLAYBOOKS_DIR
 from demisto_sdk.commands.content_graph.common import ContentType
 from demisto_sdk.commands.content_graph.objects.base_playbook import BasePlaybook
 
 
 class Playbook(BasePlaybook, content_type=ContentType.PLAYBOOK):  # type: ignore[call-arg]
     is_test: bool = False
-    tasks: dict = Field({}, exclude=True)
 
     @staticmethod
     def match(_dict: dict, path: Path) -> bool:

demisto_sdk/commands/content_graph/parsers/base_playbook.py

@@ -5,6 +5,7 @@
 import networkx
 
 from demisto_sdk.commands.common.constants import MarketplaceVersions
+from demisto_sdk.commands.common.tools import get_value
 from demisto_sdk.commands.common.update_id_set import (
     BUILT_IN_FIELDS,
     build_tasks_graph,
@@ -52,7 +53,7 @@ def __init__(
 
     @cached_property
     def field_mapping(self):
-        super().field_mapping.update({"object_id": "id"})
+        super().field_mapping.update({"object_id": "id", "tasks": "tasks"})
         return super().field_mapping
 
     def is_mandatory_dependency(self, task_id: str) -> bool:
@@ -80,6 +81,10 @@ def handle_playbook_task(self, task: Dict[str, Any], is_mandatory: bool) -> None
                 mandatorily=is_mandatory,
             )
 
+    @property
+    def tasks(self) -> Optional[Dict]:
+        return get_value(self.yml_data, self.field_mapping.get("tasks", ""), {})
+
     def handle_script_task(self, task: Dict[str, Any], is_mandatory: bool) -> None:
         """Collects a script dependency.
 

demisto_sdk/commands/content_graph/parsers/playbook.py

@@ -1,4 +1,3 @@
-from functools import cached_property
 from pathlib import Path
 from typing import List, Optional, Set
 
@@ -38,12 +37,3 @@ def supported_marketplaces(self) -> Set[MarketplaceVersions]:
             MarketplaceVersions.XSOAR_SAAS,
             MarketplaceVersions.XSOAR_ON_PREM,
         }
-
-    @cached_property
-    def field_mapping(self):
-        super().field_mapping.update({"tasks": "tasks"})
-        return super().field_mapping
-
-    @property
-    def tasks(self):
-        return self.yml_data.get("tasks", {})

demisto_sdk/commands/validate/tests/PB_validators_test.py

@@ -1,5 +1,6 @@
 import pytest
 
+from demisto_sdk.commands.content_graph.objects.base_playbook import TaskConfig
 from demisto_sdk.commands.validate.tests.test_tools import create_playbook_object
 from demisto_sdk.commands.validate.validators.PB_validators.PB100_is_no_rolename import (
     IsNoRolenameValidator,
@@ -23,35 +24,33 @@
     [
         (
             create_playbook_object(
-                paths=["inputs", "tasks"],
+                paths=["inputs", "tasks.0.task.key"],
                 values=[
                     [{"key": "input_name1", "value": "input_value1"}],
-                    {"key": {"first_input": "inputs.input_name1"}},
+                    {"first_input": "inputs.input_name1"},
                 ],
             ),
             [],
         ),
         (
             create_playbook_object(
-                paths=["inputs", "tasks"],
+                paths=["inputs", "tasks.0.task.key"],
                 values=[
                     [{"key": "input_name1", "value": "input_value1"}],
                     {
-                        "key": {
-                            "first_input": "inputs.input_name1",
-                            "second_input": "inputs.input_name2",
-                        }
+                        "first_input": "inputs.input_name1",
+                        "second_input": "inputs.input_name2",
                     },
                 ],
             ),
             [],
         ),
         (
             create_playbook_object(
-                paths=["inputs", "tasks"],
+                paths=["inputs", "tasks.0.task.key"],
                 values=[
                     [{"key": "input_name2", "value": "input_value2"}],
-                    {"key": {"first_input": "inputs.input_name1"}},
+                    {"first_input": "inputs.input_name1"},
                 ],
             ),
             "The playbook 'Detonate File - JoeSecurity V2' contains the following inputs that are not used in any of its tasks: input_name2",
@@ -192,12 +191,16 @@ def test_IsAskConditionHasUnreachableConditionValidator():
     playbook = create_playbook_object()
     assert not IsAskConditionHasUnreachableConditionValidator().is_valid([playbook])
     playbook.tasks = {
-        "0": {
-            "id": "test task",
-            "type": "condition",
-            "message": {"replyOptions": ["yes"]},
-            "nexttasks": {"no": ["1"], "yes": ["2"]},
-        }
+        "0": TaskConfig(
+            **{
+                "id": "test task",
+                "taskid": "27b9c747-b883-4878-8b60-7f352098a631",
+                "type": "condition",
+                "message": {"replyOptions": ["yes"]},
+                "nexttasks": {"no": ["1"], "yes": ["2"]},
+                "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+            }
+        )
     }
     assert IsAskConditionHasUnreachableConditionValidator().is_valid([playbook])
 
@@ -207,11 +210,15 @@ def test_IsAskConditionHasUnhandledReplyOptionsValidator():
     playbook = create_playbook_object()
     assert not IsAskConditionHasUnhandledReplyOptionsValidator().is_valid([playbook])
     playbook.tasks = {
-        "0": {
-            "id": "test task",
-            "type": "condition",
-            "message": {"replyOptions": ["yes"]},
-            "nexttasks": {"no": ["1"]},
-        }
+        "0": TaskConfig(
+            **{
+                "id": "test task",
+                "taskid": "27b9c747-b883-4878-8b60-7f352098a631",
+                "type": "condition",
+                "message": {"replyOptions": ["yes"]},
+                "nexttasks": {"no": ["1"]},
+                "task": {"id": "27b9c747-b883-4878-8b60-7f352098a63c"},
+            }
+        )
     }
     assert IsAskConditionHasUnhandledReplyOptionsValidator().is_valid([playbook])