From d47a607ea2434a9f51da04499d4ba3ddf55448bd Mon Sep 17 00:00:00 2001 From: Jay Chung Date: Thu, 8 Feb 2024 14:54:39 +0800 Subject: [PATCH 1/4] fix: missing task_execute_type for task fix: #138 --- docs/source/tasks/sub_workflow.rst | 38 ++++ src/pydolphinscheduler/core/engine.py | 4 +- src/pydolphinscheduler/core/task.py | 178 ++++++++++++++++++ .../examples/task_sub_workflow_example.py | 70 +++++++ src/pydolphinscheduler/tasks/condition.py | 4 +- src/pydolphinscheduler/tasks/datax.py | 6 +- src/pydolphinscheduler/tasks/dependent.py | 4 +- src/pydolphinscheduler/tasks/dvc.py | 4 +- src/pydolphinscheduler/tasks/http.py | 4 +- src/pydolphinscheduler/tasks/kubernetes.py | 4 +- src/pydolphinscheduler/tasks/mlflow.py | 4 +- src/pydolphinscheduler/tasks/openmldb.py | 4 +- src/pydolphinscheduler/tasks/procedure.py | 4 +- src/pydolphinscheduler/tasks/python.py | 4 +- src/pydolphinscheduler/tasks/pytorch.py | 4 +- src/pydolphinscheduler/tasks/sagemaker.py | 4 +- src/pydolphinscheduler/tasks/shell.py | 4 +- src/pydolphinscheduler/tasks/sql.py | 4 +- src/pydolphinscheduler/tasks/sub_workflow.py | 4 +- src/pydolphinscheduler/tasks/switch.py | 4 +- tests/core/test_engine.py | 1 + tests/testing/constants.py | 1 - 22 files changed, 322 insertions(+), 36 deletions(-) create mode 100644 src/pydolphinscheduler/examples/task_sub_workflow_example.py diff --git a/docs/source/tasks/sub_workflow.rst b/docs/source/tasks/sub_workflow.rst index 026131af..a33de839 100644 --- a/docs/source/tasks/sub_workflow.rst +++ b/docs/source/tasks/sub_workflow.rst @@ -18,6 +18,44 @@ Sub Workflow ============ +Task trigger exists workflow run, should make sure workflow exists in current project when you create +sub workflow task. + +Example +------- + +we have a simple example about how to use sub workflow task, when we want to create a sub workflow task, +we should makeh sure in already exists in current project. So the first thing we do is to create a workflow +will be used as sub workflow task. + +.. literalinclude:: ../../../src/pydolphinscheduler/examples/sub_workflow_example.py + :start-after: [start sub_workflow_declare] + :end-before: [end sub_workflow_declare] + +workflow with name ``sub_workflow_upstream`` would be create after we exists ``submit`` method. + +Then we create a main workflow, and the sub workflow task will connect to workflow we created before. + +.. literalinclude:: ../../../src/pydolphinscheduler/examples/sub_workflow_example.py + :start-after: [start sub_workflow_task_declare] + :end-before: [end sub_workflow_task_declare] + +Finish we can submit or run sub workflow task by ``submit`` or ``run`` method. And you can also use workflow +already exists in current project instead of create a new one. + +.. note:: + + We could only run the workflow contains sub workflow task, and the sub workflow task will trigger the + sub workflow run. + + +.. literalinclude:: ../../../src/pydolphinscheduler/examples/task_sub_workflow_example.py + :start-after: [start workflow_declare] + :end-before: [end workflow_declare] + +Dive Into +--------- + .. automodule:: pydolphinscheduler.tasks.sub_workflow diff --git a/src/pydolphinscheduler/core/engine.py b/src/pydolphinscheduler/core/engine.py index 3e36820d..b5bf0c80 100644 --- a/src/pydolphinscheduler/core/engine.py +++ b/src/pydolphinscheduler/core/engine.py @@ -21,7 +21,7 @@ from py4j.protocol import Py4JJavaError -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.exceptions import PyDSParamException from pydolphinscheduler.java_gateway import gateway @@ -34,7 +34,7 @@ class ProgramType(str): PYTHON = "PYTHON" -class Engine(Task): +class Engine(BatchTask): """Task engine object, declare behavior for engine task to dolphinscheduler. This is the parent class of spark, flink and mr tasks, diff --git a/src/pydolphinscheduler/core/task.py b/src/pydolphinscheduler/core/task.py index ac48ab83..a8e5fac4 100644 --- a/src/pydolphinscheduler/core/task.py +++ b/src/pydolphinscheduler/core/task.py @@ -511,3 +511,181 @@ def add_out( """ self._output_params[name] = value + + +class BatchTask(Task): + """Task object, parent class for all exactly task type. + + :param name: The name of the task. Node names within the same workflow must be unique. + :param task_type: + :param description: default None + :param flag: default TaskFlag.YES, + :param task_priority: default TaskPriority.MEDIUM + :param worker_group: default configuration.WORKFLOW_WORKER_GROUP + :param environment_name: default None + :param task_group_id: Identify of task group to restrict the parallelism of tasks instance run, default 0. + :param task_group_priority: Priority for same task group to, the higher the value, the higher the + priority, default 0. + :param delay_time: deault 0 + :param fail_retry_times: default 0 + :param fail_retry_interval: default 1 + :param timeout_notify_strategy: default, None + :param timeout: Timeout attribute for task, in minutes. Task is consider as timed out task when the + running time of a task exceeds than this value. when data type is :class:`datetime.timedelta` will + be converted to int(in minutes). default ``None`` + :param resource_list: default None + :param wait_start_timeout: default None + :param condition_result: default None, + :param resource_plugin: default None + :param is_cache: default False + :param input_params: default None, input parameters, {param_name: param_value} + :param output_params: default None, input parameters, {param_name: param_value} + """ + + _DEFINE_ATTR = Task._DEFINE_ATTR | {"task_execute_type"} + + def __init__( + self, + name: str, + task_type: str, + description: str | None = None, + flag: str | None = TaskFlag.YES, + task_priority: str | None = TaskPriority.MEDIUM, + worker_group: str | None = configuration.WORKFLOW_WORKER_GROUP, + environment_name: str | None = None, + task_group_id: int | None = 0, + task_group_priority: int | None = 0, + delay_time: int | None = 0, + fail_retry_times: int | None = 0, + fail_retry_interval: int | None = 1, + timeout_notify_strategy: str | None = None, + timeout: timedelta | int | None = None, + workflow: Workflow | None = None, + resource_list: list | None = None, + dependence: dict | None = None, + wait_start_timeout: dict | None = None, + condition_result: dict | None = None, + resource_plugin: ResourcePlugin | None = None, + is_cache: bool | None = False, + input_params: dict | None = None, + output_params: dict | None = None, + *args, + **kwargs, + ): + super().__init__( + name, + task_type, + description, + flag, + task_priority, + worker_group, + environment_name, + task_group_id, + task_group_priority, + delay_time, + fail_retry_times, + fail_retry_interval, + timeout_notify_strategy, + timeout, + workflow, + resource_list, + dependence, + wait_start_timeout, + condition_result, + resource_plugin, + is_cache, + input_params, + output_params, + *args, + **kwargs, + ) + self.task_execute_type = "BATCH" + + +class StreamTask(Task): + """Task object, parent class for all exactly task type. + + :param name: The name of the task. Node names within the same workflow must be unique. + :param task_type: + :param description: default None + :param flag: default TaskFlag.YES, + :param task_priority: default TaskPriority.MEDIUM + :param worker_group: default configuration.WORKFLOW_WORKER_GROUP + :param environment_name: default None + :param task_group_id: Identify of task group to restrict the parallelism of tasks instance run, default 0. + :param task_group_priority: Priority for same task group to, the higher the value, the higher the + priority, default 0. + :param delay_time: deault 0 + :param fail_retry_times: default 0 + :param fail_retry_interval: default 1 + :param timeout_notify_strategy: default, None + :param timeout: Timeout attribute for task, in minutes. Task is consider as timed out task when the + running time of a task exceeds than this value. when data type is :class:`datetime.timedelta` will + be converted to int(in minutes). default ``None`` + :param resource_list: default None + :param wait_start_timeout: default None + :param condition_result: default None, + :param resource_plugin: default None + :param is_cache: default False + :param input_params: default None, input parameters, {param_name: param_value} + :param output_params: default None, input parameters, {param_name: param_value} + """ + + _DEFINE_ATTR = Task._DEFINE_ATTR | {"task_execute_type"} + + def __init__( + self, + name: str, + task_type: str, + description: str | None = None, + flag: str | None = TaskFlag.YES, + task_priority: str | None = TaskPriority.MEDIUM, + worker_group: str | None = configuration.WORKFLOW_WORKER_GROUP, + environment_name: str | None = None, + task_group_id: int | None = 0, + task_group_priority: int | None = 0, + delay_time: int | None = 0, + fail_retry_times: int | None = 0, + fail_retry_interval: int | None = 1, + timeout_notify_strategy: str | None = None, + timeout: timedelta | int | None = None, + workflow: Workflow | None = None, + resource_list: list | None = None, + dependence: dict | None = None, + wait_start_timeout: dict | None = None, + condition_result: dict | None = None, + resource_plugin: ResourcePlugin | None = None, + is_cache: bool | None = False, + input_params: dict | None = None, + output_params: dict | None = None, + *args, + **kwargs, + ): + super().__init__( + name, + task_type, + description, + flag, + task_priority, + worker_group, + environment_name, + task_group_id, + task_group_priority, + delay_time, + fail_retry_times, + fail_retry_interval, + timeout_notify_strategy, + timeout, + workflow, + resource_list, + dependence, + wait_start_timeout, + condition_result, + resource_plugin, + is_cache, + input_params, + output_params, + *args, + **kwargs, + ) + self.task_execute_type = "STREAM" diff --git a/src/pydolphinscheduler/examples/task_sub_workflow_example.py b/src/pydolphinscheduler/examples/task_sub_workflow_example.py new file mode 100644 index 00000000..a9a06c51 --- /dev/null +++ b/src/pydolphinscheduler/examples/task_sub_workflow_example.py @@ -0,0 +1,70 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +r"""A tutorial example take you to experience pydolphinscheduler. + +After tutorial.py file submit to Apache DolphinScheduler server a DAG would be create, +and workflow DAG graph as below: + + --> task_child_one + / \ +task_parent --> --> task_union + \ / + --> task_child_two + +it will instantiate and run all the task it have. +""" + +# [start tutorial] +# [start package_import] +# Import Workflow object to define your workflow attributes +from pydolphinscheduler.core.workflow import Workflow + +# Import task Shell object cause we would create some shell tasks later +from pydolphinscheduler.tasks.sub_workflow import SubWorkflow +from pydolphinscheduler.tasks.shell import Shell + + +# [start workflow_declare] +# [start sub_workflow_declare] +with Workflow(name="sub_workflow_downstream") as wf_downstream, Workflow( + name="task_sub_workflow_example" +) as wf_upstream: + sub_workflow_ds_task = Shell( + name="task_sub_workflow", + command="echo 'call sub workflow success!'", + workflow=wf_downstream, + ) + wf_downstream.submit() + # [end sub_workflow_declare] + + sub_workflow_pre = Shell( + name="pre-task", + command="echo 'prefix task for sub workflow'", + workflow=wf_upstream, + ) + # [start sub_workflow_task_declare] + sw_task = SubWorkflow( + name="sub_workflow", + workflow_name=wf_downstream.name, + workflow=wf_upstream, + ) + # [end sub_workflow_task_declare] + sub_workflow_pre >> sw_task + # Please make sure workflow with name `wf_downstream.name` exists when we submit or run sub workflow task + wf_upstream.run() +# [end workflow_declare] diff --git a/src/pydolphinscheduler/tasks/condition.py b/src/pydolphinscheduler/tasks/condition.py index 3d8f6f8d..9359b5d0 100644 --- a/src/pydolphinscheduler/tasks/condition.py +++ b/src/pydolphinscheduler/tasks/condition.py @@ -20,7 +20,7 @@ from __future__ import annotations from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask, Task from pydolphinscheduler.exceptions import PyDSParamException from pydolphinscheduler.models.base import Base @@ -154,7 +154,7 @@ def __init__(self, *args): super().__init__(*args) -class Condition(Task): +class Condition(BatchTask): """Task condition object, declare behavior for condition task to dolphinscheduler.""" def __init__( diff --git a/src/pydolphinscheduler/tasks/datax.py b/src/pydolphinscheduler/tasks/datax.py index 59edfee2..c6734ce6 100644 --- a/src/pydolphinscheduler/tasks/datax.py +++ b/src/pydolphinscheduler/tasks/datax.py @@ -21,11 +21,11 @@ from pydolphinscheduler.constants import TaskType from pydolphinscheduler.core.mixin import WorkerResourceMixin -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.models.datasource import Datasource -class CustomDataX(WorkerResourceMixin, Task): +class CustomDataX(WorkerResourceMixin, BatchTask): """Task CustomDatax object, declare behavior for custom DataX task to dolphinscheduler. You provider json template for DataX, it can synchronize data according to the template you provided. @@ -87,7 +87,7 @@ def __init__( self.add_attr(**kwargs) -class DataX(WorkerResourceMixin, Task): +class DataX(WorkerResourceMixin, BatchTask): """Task DataX object, declare behavior for DataX task to dolphinscheduler. It should run database datax job in multiply sql link engine, such as: diff --git a/src/pydolphinscheduler/tasks/dependent.py b/src/pydolphinscheduler/tasks/dependent.py index 6c5ac2b5..be716f65 100644 --- a/src/pydolphinscheduler/tasks/dependent.py +++ b/src/pydolphinscheduler/tasks/dependent.py @@ -21,7 +21,7 @@ import warnings from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.exceptions import PyDSJavaGatewayException, PyDSParamException from pydolphinscheduler.java_gateway import gateway from pydolphinscheduler.models.base import Base @@ -272,7 +272,7 @@ def __init__(self, *args): super().__init__(*args) -class Dependent(Task): +class Dependent(BatchTask): """Task dependent object, declare behavior for dependent task to dolphinscheduler.""" def __init__(self, name: str, dependence: DependentOperator, *args, **kwargs): diff --git a/src/pydolphinscheduler/tasks/dvc.py b/src/pydolphinscheduler/tasks/dvc.py index 5cbc6f01..46b21259 100644 --- a/src/pydolphinscheduler/tasks/dvc.py +++ b/src/pydolphinscheduler/tasks/dvc.py @@ -21,7 +21,7 @@ from copy import deepcopy from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask class DvcTaskType(str): @@ -32,7 +32,7 @@ class DvcTaskType(str): UPLOAD = "Upload" -class BaseDVC(Task): +class BaseDVC(BatchTask): """Base class for dvc task.""" dvc_task_type = None diff --git a/src/pydolphinscheduler/tasks/http.py b/src/pydolphinscheduler/tasks/http.py index e64494bc..000cc26f 100644 --- a/src/pydolphinscheduler/tasks/http.py +++ b/src/pydolphinscheduler/tasks/http.py @@ -23,7 +23,7 @@ from pydolphinscheduler.constants import TaskType from pydolphinscheduler.core.parameter import Direction, ParameterHelper -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.exceptions import PyDSParamException @@ -53,7 +53,7 @@ class HttpCheckCondition: BODY_NOT_CONTAINS = "BODY_NOT_CONTAINS" -class Http(Task): +class Http(BatchTask): """Task HTTP object, declare behavior for HTTP task to dolphinscheduler. :param name: The name or identifier for the HTTP task. diff --git a/src/pydolphinscheduler/tasks/kubernetes.py b/src/pydolphinscheduler/tasks/kubernetes.py index b9029278..0152cb2f 100644 --- a/src/pydolphinscheduler/tasks/kubernetes.py +++ b/src/pydolphinscheduler/tasks/kubernetes.py @@ -17,10 +17,10 @@ """Task Kubernetes.""" from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask -class Kubernetes(Task): +class Kubernetes(BatchTask): """Task Kubernetes object, declare behavior for Kubernetes task to dolphinscheduler. :param name: task name diff --git a/src/pydolphinscheduler/tasks/mlflow.py b/src/pydolphinscheduler/tasks/mlflow.py index f61fc863..2d03709a 100644 --- a/src/pydolphinscheduler/tasks/mlflow.py +++ b/src/pydolphinscheduler/tasks/mlflow.py @@ -21,7 +21,7 @@ from copy import deepcopy from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask class MLflowTaskType(str): @@ -50,7 +50,7 @@ class MLflowDeployType(str): DEFAULT_VERSION = "master" -class BaseMLflow(Task): +class BaseMLflow(BatchTask): """Base MLflow task.""" mlflow_task_type = None diff --git a/src/pydolphinscheduler/tasks/openmldb.py b/src/pydolphinscheduler/tasks/openmldb.py index 5dad36ec..bb22993a 100644 --- a/src/pydolphinscheduler/tasks/openmldb.py +++ b/src/pydolphinscheduler/tasks/openmldb.py @@ -18,10 +18,10 @@ """Task OpenMLDB.""" from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask -class OpenMLDB(Task): +class OpenMLDB(BatchTask): """Task OpenMLDB object, declare behavior for OpenMLDB task to dolphinscheduler. :param name: task name diff --git a/src/pydolphinscheduler/tasks/procedure.py b/src/pydolphinscheduler/tasks/procedure.py index 807196af..fd22cd34 100644 --- a/src/pydolphinscheduler/tasks/procedure.py +++ b/src/pydolphinscheduler/tasks/procedure.py @@ -20,11 +20,11 @@ from __future__ import annotations from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.models.datasource import Datasource -class Procedure(Task): +class Procedure(BatchTask): """Task Procedure object, declare behavior for Procedure task to dolphinscheduler. It should run database procedure job in multiply sql lik engine, such as: diff --git a/src/pydolphinscheduler/tasks/python.py b/src/pydolphinscheduler/tasks/python.py index 8e2729f7..80dced73 100644 --- a/src/pydolphinscheduler/tasks/python.py +++ b/src/pydolphinscheduler/tasks/python.py @@ -28,13 +28,13 @@ from pydolphinscheduler.constants import TaskType from pydolphinscheduler.core.mixin import WorkerResourceMixin -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.exceptions import PyDSParamException log = logging.getLogger(__file__) -class Python(WorkerResourceMixin, Task): +class Python(WorkerResourceMixin, BatchTask): """Task Python object, declare behavior for Python task to dolphinscheduler. Python task support two types of parameters for :param:``definition``, and here is an example: diff --git a/src/pydolphinscheduler/tasks/pytorch.py b/src/pydolphinscheduler/tasks/pytorch.py index 0ba0d7d8..db0d9be5 100644 --- a/src/pydolphinscheduler/tasks/pytorch.py +++ b/src/pydolphinscheduler/tasks/pytorch.py @@ -20,7 +20,7 @@ from pydolphinscheduler.constants import TaskType from pydolphinscheduler.core.mixin import WorkerResourceMixin -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask class DEFAULT: @@ -31,7 +31,7 @@ class DEFAULT: python_command = "${PYTHON_HOME}" -class Pytorch(WorkerResourceMixin, Task): +class Pytorch(WorkerResourceMixin, BatchTask): """Task Pytorch object, declare behavior for Pytorch task to dolphinscheduler. See also: `DolphinScheduler Pytorch Task Plugin diff --git a/src/pydolphinscheduler/tasks/sagemaker.py b/src/pydolphinscheduler/tasks/sagemaker.py index 30b128d1..01c8fac4 100644 --- a/src/pydolphinscheduler/tasks/sagemaker.py +++ b/src/pydolphinscheduler/tasks/sagemaker.py @@ -18,10 +18,10 @@ """Task SageMaker.""" from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask -class SageMaker(Task): +class SageMaker(BatchTask): """Task SageMaker object, declare behavior for SageMaker task to dolphinscheduler. :param name: A unique, meaningful string for the SageMaker task. diff --git a/src/pydolphinscheduler/tasks/shell.py b/src/pydolphinscheduler/tasks/shell.py index f6795ca2..c93980b1 100644 --- a/src/pydolphinscheduler/tasks/shell.py +++ b/src/pydolphinscheduler/tasks/shell.py @@ -19,10 +19,10 @@ from pydolphinscheduler.constants import TaskType from pydolphinscheduler.core.mixin import WorkerResourceMixin -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask -class Shell(WorkerResourceMixin, Task): +class Shell(WorkerResourceMixin, BatchTask): """Task shell object, declare behavior for shell task to dolphinscheduler. :param name: A unique, meaningful string for the shell task. diff --git a/src/pydolphinscheduler/tasks/sql.py b/src/pydolphinscheduler/tasks/sql.py index 520c2917..c9c3f490 100644 --- a/src/pydolphinscheduler/tasks/sql.py +++ b/src/pydolphinscheduler/tasks/sql.py @@ -23,7 +23,7 @@ from collections.abc import Sequence from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.models.datasource import Datasource log = logging.getLogger(__file__) @@ -36,7 +36,7 @@ class SqlType: NOT_SELECT = "1" -class Sql(Task): +class Sql(BatchTask): """Task SQL object, declare behavior for SQL task to dolphinscheduler. It should run sql job in multiply sql lik engine, such as: diff --git a/src/pydolphinscheduler/tasks/sub_workflow.py b/src/pydolphinscheduler/tasks/sub_workflow.py index 715a21e5..46711463 100644 --- a/src/pydolphinscheduler/tasks/sub_workflow.py +++ b/src/pydolphinscheduler/tasks/sub_workflow.py @@ -20,12 +20,12 @@ from __future__ import annotations from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask from pydolphinscheduler.exceptions import PyDSWorkflowNotAssignException from pydolphinscheduler.java_gateway import gateway -class SubWorkflow(Task): +class SubWorkflow(BatchTask): """Task SubWorkflow object, declare behavior for SubWorkflow task to dolphinscheduler.""" _task_custom_attr = {"process_definition_code"} diff --git a/src/pydolphinscheduler/tasks/switch.py b/src/pydolphinscheduler/tasks/switch.py index b0969afc..625ac702 100644 --- a/src/pydolphinscheduler/tasks/switch.py +++ b/src/pydolphinscheduler/tasks/switch.py @@ -19,7 +19,7 @@ from __future__ import annotations from pydolphinscheduler.constants import TaskType -from pydolphinscheduler.core.task import Task +from pydolphinscheduler.core.task import BatchTask, Task from pydolphinscheduler.exceptions import PyDSParamException from pydolphinscheduler.models.base import Base @@ -126,7 +126,7 @@ def get_define(self, camel_attr=True) -> dict: return super().get_define() -class Switch(Task): +class Switch(BatchTask): """Task switch object, declare behavior for switch task to dolphinscheduler. Param of workflow or at least one local param of task must be set diff --git a/tests/core/test_engine.py b/tests/core/test_engine.py index 9c443084..312f27a5 100644 --- a/tests/core/test_engine.py +++ b/tests/core/test_engine.py @@ -133,6 +133,7 @@ def test_property_task_params(mock_resource, mock_code_version, attr, expect): "timeoutFlag": "CLOSE", "timeoutNotifyStrategy": None, "timeout": 0, + "taskExecuteType": "BATCH", }, ) ], diff --git a/tests/testing/constants.py b/tests/testing/constants.py index 07a32494..3b106c1f 100644 --- a/tests/testing/constants.py +++ b/tests/testing/constants.py @@ -23,7 +23,6 @@ # but most of them just without adding by mistake, and we should add it later. task_without_example = { "http", - "sub_workflow", "procedure", } From b472e67c5ac56477dfedd3a7385a517bde39fa53 Mon Sep 17 00:00:00 2001 From: Jay Chung Date: Thu, 8 Feb 2024 14:56:35 +0800 Subject: [PATCH 2/4] correct docstring --- .../examples/task_sub_workflow_example.py | 14 +------------- 1 file changed, 1 insertion(+), 13 deletions(-) diff --git a/src/pydolphinscheduler/examples/task_sub_workflow_example.py b/src/pydolphinscheduler/examples/task_sub_workflow_example.py index a9a06c51..df7b6ce4 100644 --- a/src/pydolphinscheduler/examples/task_sub_workflow_example.py +++ b/src/pydolphinscheduler/examples/task_sub_workflow_example.py @@ -15,19 +15,7 @@ # specific language governing permissions and limitations # under the License. -r"""A tutorial example take you to experience pydolphinscheduler. - -After tutorial.py file submit to Apache DolphinScheduler server a DAG would be create, -and workflow DAG graph as below: - - --> task_child_one - / \ -task_parent --> --> task_union - \ / - --> task_child_two - -it will instantiate and run all the task it have. -""" +"""A example workflow for task sub workflow.""" # [start tutorial] # [start package_import] From 8ff5f8b3bffb603b8bb3b6ffb14de687834a4b6f Mon Sep 17 00:00:00 2001 From: Jay Chung Date: Thu, 8 Feb 2024 15:03:23 +0800 Subject: [PATCH 3/4] Add sub workflow to index.rst --- docs/source/tasks/index.rst | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/source/tasks/index.rst b/docs/source/tasks/index.rst index fa2e73a0..96e57da4 100644 --- a/docs/source/tasks/index.rst +++ b/docs/source/tasks/index.rst @@ -28,6 +28,7 @@ In this section sql python http + sub_workflow switch condition From 6842efac9c70bd918acedeb597c300cf340b6726 Mon Sep 17 00:00:00 2001 From: Jay Chung Date: Thu, 8 Feb 2024 15:08:03 +0800 Subject: [PATCH 4/4] fix docs --- docs/source/tasks/index.rst | 2 -- docs/source/tasks/sub_workflow.rst | 4 ++-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/docs/source/tasks/index.rst b/docs/source/tasks/index.rst index 96e57da4..153d7abb 100644 --- a/docs/source/tasks/index.rst +++ b/docs/source/tasks/index.rst @@ -41,8 +41,6 @@ In this section kubernetes datax - sub_workflow - sagemaker mlflow openmldb diff --git a/docs/source/tasks/sub_workflow.rst b/docs/source/tasks/sub_workflow.rst index a33de839..eca01685 100644 --- a/docs/source/tasks/sub_workflow.rst +++ b/docs/source/tasks/sub_workflow.rst @@ -28,7 +28,7 @@ we have a simple example about how to use sub workflow task, when we want to cre we should makeh sure in already exists in current project. So the first thing we do is to create a workflow will be used as sub workflow task. -.. literalinclude:: ../../../src/pydolphinscheduler/examples/sub_workflow_example.py +.. literalinclude:: ../../../src/pydolphinscheduler/examples/task_sub_workflow_example.py :start-after: [start sub_workflow_declare] :end-before: [end sub_workflow_declare] @@ -36,7 +36,7 @@ workflow with name ``sub_workflow_upstream`` would be create after we exists ``s Then we create a main workflow, and the sub workflow task will connect to workflow we created before. -.. literalinclude:: ../../../src/pydolphinscheduler/examples/sub_workflow_example.py +.. literalinclude:: ../../../src/pydolphinscheduler/examples/task_sub_workflow_example.py :start-after: [start sub_workflow_task_declare] :end-before: [end sub_workflow_task_declare]