This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch main
in repository
https://gitbox.apache.org/repos/asf/dolphinscheduler-sdk-python.git
The following commit(s) were added to refs/heads/main by this push:
new d47f03c [impv] Attribute timeout of task and workflow accept both
timedelta and int (#123)
d47f03c is described below
commit d47f03cbfbfdb9e4c61c91be5d32a9956d080171
Author: Luke Yan <[email protected]>
AuthorDate: Mon Dec 25 12:04:59 2023 +0800
[impv] Attribute timeout of task and workflow accept both timedelta and int
(#123)
---------
Co-authored-by: lukeyan <[email protected]>
Co-authored-by: Jay Chung <[email protected]>
---
src/pydolphinscheduler/core/task.py | 17 +++++++++--------
src/pydolphinscheduler/core/workflow.py | 26 +++++++++++++++++++++-----
tests/core/test_task.py | 8 ++++++--
tests/core/test_workflow.py | 23 ++++++++++++++++++++---
4 files changed, 56 insertions(+), 18 deletions(-)
diff --git a/src/pydolphinscheduler/core/task.py
b/src/pydolphinscheduler/core/task.py
index 9f6b644..5393322 100644
--- a/src/pydolphinscheduler/core/task.py
+++ b/src/pydolphinscheduler/core/task.py
@@ -99,7 +99,9 @@ class Task(Base):
:param fail_retry_times: default 0
:param fail_retry_interval: default 1
:param timeout_notify_strategy: default, None
- :param timeout: 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,
@@ -164,7 +166,7 @@ class Task(Base):
fail_retry_times: Optional[int] = 0,
fail_retry_interval: Optional[int] = 1,
timeout_notify_strategy: Optional = None,
- timeout: Optional[timedelta] = None,
+ timeout: Optional[Union[timedelta, int]] = None,
workflow: Optional[Workflow] = None,
resource_list: Optional[List] = None,
dependence: Optional[Dict] = None,
@@ -190,7 +192,7 @@ class Task(Base):
self.fail_retry_interval = fail_retry_interval
self.delay_time = delay_time
self.timeout_notify_strategy = timeout_notify_strategy
- self._timeout: timedelta = timeout
+ self._timeout: Union[timedelta, int] = timeout
self._workflow = None
self._input_params = input_params or {}
self._output_params = output_params or {}
@@ -248,13 +250,12 @@ class Task(Base):
@property
def timeout(self) -> int:
"""Get attribute timeout."""
+ if isinstance(self._timeout, int):
+ if self._timeout < 0:
+ raise PyDSParamException("The timeout value must be greater
than 0")
+ return self._timeout
return timedelta2timeout(self._timeout) if self._timeout else 0
- @timeout.setter
- def timeout(self, val: timedelta) -> None:
- """Set attribute timeout."""
- self._timeout = val
-
@property
def timeout_flag(self) -> str:
"""Whether the timeout attribute is being set or not."""
diff --git a/src/pydolphinscheduler/core/workflow.py
b/src/pydolphinscheduler/core/workflow.py
index 582faa6..99d14ac 100644
--- a/src/pydolphinscheduler/core/workflow.py
+++ b/src/pydolphinscheduler/core/workflow.py
@@ -18,7 +18,7 @@
"""Module workflow, core class for workflow define."""
import json
-from datetime import datetime
+from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Set, Union
from pydolphinscheduler import configuration
@@ -28,7 +28,12 @@ from pydolphinscheduler.core.resource_plugin import
ResourcePlugin
from pydolphinscheduler.exceptions import PyDSParamException,
PyDSTaskNoFoundException
from pydolphinscheduler.java_gateway import gateway
from pydolphinscheduler.models import Base, Project, User
-from pydolphinscheduler.utils.date import MAX_DATETIME, conv_from_str,
conv_to_schedule
+from pydolphinscheduler.utils.date import (
+ MAX_DATETIME,
+ conv_from_str,
+ conv_to_schedule,
+ timedelta2timeout,
+)
class WorkflowContext:
@@ -75,7 +80,9 @@ class Workflow(Base):
finished.
* ``SERIAL_PRIORITY``: means the all instance will wait for the
previous instance to finish, and
all the waiting instances will be executed base on workflow
priority order.
-
+ :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 ``0``
:param user: The user for current workflow. Will create a new one if it do
not exists. If your
parameter ``project`` already exists but project's create do not
belongs to ``user``, will grant
``project`` to ``user`` automatically.
@@ -130,7 +137,7 @@ class Workflow(Base):
warning_type: Optional[str] = configuration.WORKFLOW_WARNING_TYPE,
warning_group_id: Optional[int] = 0,
execution_type: Optional[str] = configuration.WORKFLOW_EXECUTION_TYPE,
- timeout: Optional[int] = 0,
+ timeout: Optional[Union[timedelta, int]] = 0,
release_state: Optional[str] = configuration.WORKFLOW_RELEASE_STATE,
param: Optional[Dict] = None,
resource_plugin: Optional[ResourcePlugin] = None,
@@ -180,7 +187,7 @@ class Workflow(Base):
)
else:
self._execution_type = execution_type
- self.timeout = timeout
+ self._timeout: Union[timedelta, int] = timeout
self._release_state = release_state
self.param = param
self.tasks: dict = {}
@@ -263,6 +270,15 @@ class Workflow(Base):
"""Set attribute release_state."""
self._release_state = val.lower()
+ @property
+ def timeout(self) -> int:
+ """Get attribute timeout."""
+ if isinstance(self._timeout, int):
+ if self._timeout < 0:
+ raise PyDSParamException("The timeout value must be greater
than 0")
+ return self._timeout
+ return timedelta2timeout(self._timeout) if self._timeout else 0
+
@property
def execution_type(self) -> str:
"""Get attribute execution_type."""
diff --git a/tests/core/test_task.py b/tests/core/test_task.py
index 1a44e6b..3e62157 100644
--- a/tests/core/test_task.py
+++ b/tests/core/test_task.py
@@ -20,7 +20,7 @@ import logging
import re
import warnings
from datetime import timedelta
-from typing import Set, Tuple
+from typing import Set, Tuple, Union
from unittest.mock import PropertyMock, patch
import pytest
@@ -111,9 +111,13 @@ def test__get_attr(addition: Set, ignore: Set, expect:
Set):
(timedelta(seconds=61), (2, "OPEN")),
(timedelta(seconds=0), (0, "CLOSE")),
(timedelta(minutes=1.3), (2, "OPEN")),
+ (1, (1, "OPEN")),
+ (5, (5, "OPEN")),
+ (0, (0, "CLOSE")),
+ (None, (0, "CLOSE")),
],
)
-def test_task_timeout(value: timedelta, expect: Tuple[int, str]):
+def test_task_timeout(value: Union[timedelta, int], expect: Tuple[int, str]):
"""Test task timout attribute."""
task = TestTask(
name="test-get-attr",
diff --git a/tests/core/test_workflow.py b/tests/core/test_workflow.py
index d3f83f5..3e9fbd0 100644
--- a/tests/core/test_workflow.py
+++ b/tests/core/test_workflow.py
@@ -17,8 +17,8 @@
"""Test workflow."""
import warnings
-from datetime import datetime
-from typing import Any, List
+from datetime import datetime, timedelta
+from typing import Any, List, Union
from unittest.mock import patch
import pytest
@@ -90,7 +90,6 @@ def test_workflow_default_value(name, value):
("warning_type", str, "FAILURE"),
("warning_group_id", int, 1),
("execution_type", str, "PARALLEL"),
- ("timeout", int, 1),
("param", dict, {"key": "value"}),
(
"resource_list",
@@ -108,6 +107,24 @@ def test_set_attr(name, cls, expect):
), f"Workflow set attribute `{name}` do not work expect"
[email protected](
+ "value, expect",
+ [
+ (0, 0),
+ (5, 5),
+ (timedelta(seconds=60), 1),
+ (timedelta(seconds=61), 2),
+ (timedelta(seconds=360), 6),
+ ],
+)
+def test_workflow_timeout(value: Union[timedelta, int], expect: int):
+ """Test workflow timout attribute."""
+ with Workflow(TEST_WORKFLOW_NAME, timeout=value) as workflow:
+ assert (
+ workflow.timeout == expect
+ ), f"Workflow set attribute timeout expect {expect} but get
{workflow.timeout}"
+
+
@pytest.mark.parametrize(
"value,expect",
[
