Skip to content

Commit

Permalink
lint
Browse files Browse the repository at this point in the history
  • Loading branch information
VerstraeteBert committed Sep 2, 2024
1 parent 2d185a3 commit ae17d04
Showing 1 changed file with 11 additions and 8 deletions.
19 changes: 11 additions & 8 deletions cognite/client/data_classes/workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -314,6 +314,7 @@ def dump(self, camel_case: bool = True) -> dict[str, Any]:
("cdfRequest" if camel_case else "cdf_request"): output,
}


class SubworkflowTaskParameters(WorkflowTaskParameters):
"""
The subworkflow task parameters are used to specify a subworkflow task.
Expand All @@ -324,14 +325,19 @@ class SubworkflowTaskParameters(WorkflowTaskParameters):
- a reference to another workflow which will be loaded in at workflow start time (defined by an external ID and version).
Args:
tasks (list[WorkflowTask], optional): The tasks belonging to the subworkflow.
workflow_external_id (str, optional): The external ID of the referenced workflow.
version (str, optional): The version of the referenced workflow.
tasks (list[WorkflowTask] | None): The tasks belonging to the subworkflow.
workflow_external_id (str | None): The external ID of the referenced workflow.
version (str | None): The version of the referenced workflow.
"""

task_type = "subworkflow"

def __init__(self, tasks: list[WorkflowTask] = None, workflow_external_id: str = None, version: str = None) -> None:
def __init__(
self,
tasks: list[WorkflowTask] | None = None,
workflow_external_id: str | None = None,
version: str | None = None,
) -> None:
if tasks:
self.tasks = tasks
self.workflow_external_id = None
Expand All @@ -352,10 +358,7 @@ def _load(cls: type[Self], resource: dict, cognite_client: CogniteClient | None
tasks=[WorkflowTask._load(task) for task in subworkflow["tasks"]],
)
elif "workflowExternalId" in subworkflow and "version" in subworkflow:
return cls(
workflow_external_id=subworkflow["workflowExternalId"],
version=subworkflow["version"]
)
return cls(workflow_external_id=subworkflow["workflowExternalId"], version=subworkflow["version"])
else:
raise ValueError("Invalid subworkflow parameters provided.")

Expand Down

0 comments on commit ae17d04

Please sign in to comment.