Dependent

A dependent task type’s example and dive into information of PyDolphinScheduler.

Example

r"""
A example workflow for task dependent.

This example will create two workflows named `task_dependent` and `task_dependent_external`.
`task_dependent` is true workflow define and run task dependent, while `task_dependent_external`
define outside workflow and task from dependent.

After this script submit, we would get workflow as below:

task_dependent_external:

task_1
task_2
task_3

task_dependent:

task_dependent(this task dependent on task_dependent_external.task_1 and task_dependent_external.task_2).
"""
from pydolphinscheduler import configuration
from pydolphinscheduler.core.process_definition import ProcessDefinition
from pydolphinscheduler.tasks.dependent import And, Dependent, DependentItem, Or
from pydolphinscheduler.tasks.shell import Shell

with ProcessDefinition(
    name="task_dependent_external",
    tenant="tenant_exists",
) as pd:
    task_1 = Shell(name="task_1", command="echo task 1")
    task_2 = Shell(name="task_2", command="echo task 2")
    task_3 = Shell(name="task_3", command="echo task 3")
    pd.submit()

with ProcessDefinition(
    name="task_dependent_example",
    tenant="tenant_exists",
) as pd:
    task = Dependent(
        name="task_dependent",
        dependence=And(
            Or(
                DependentItem(
                    project_name=configuration.WORKFLOW_PROJECT,
                    process_definition_name="task_dependent_external",
                    dependent_task_name="task_1",
                ),
                DependentItem(
                    project_name=configuration.WORKFLOW_PROJECT,
                    process_definition_name="task_dependent_external",
                    dependent_task_name="task_2",
                ),
            )
        ),
    )
    pd.submit()

Dive Into

Task dependent.

class pydolphinscheduler.tasks.dependent.And(*args)

Bases: DependentOperator

Operator And for task dependent.

It could accept both DependentItem and children of DependentOperator, and set AND condition to those args.

class pydolphinscheduler.tasks.dependent.Dependent(name: str, dependence: DependentOperator, *args, **kwargs)

Bases: Task

Task dependent object, declare behavior for dependent task to dolphinscheduler.

_downstream_task_codes: Set[int]

_task_relation: Set[TaskRelation]

_upstream_task_codes: Set[int]

property task_params: Dict

Override Task.task_params for dependent task.

Dependent task have some specials attribute dependence, and in most of the task this attribute is None and use empty dict {} as default value. We do not use class attribute _task_custom_attr due to avoid attribute cover.

class pydolphinscheduler.tasks.dependent.DependentDate

Bases: str

Constant of Dependent date value.

These values set according to Java server models, if you want to add and change it, please change Java server models first.

CURRENT_HOUR = 'currentHour'

LAST_FRIDAY = 'lastFriday'

LAST_MONDAY = 'lastMonday'

LAST_MONTH = 'lastMonth'

LAST_MONTH_BEGIN = 'lastMonthBegin'

LAST_MONTH_END = 'lastMonthEnd'

LAST_ONE_DAYS = 'last1Days'

LAST_ONE_HOUR = 'last1Hour'

LAST_SATURDAY = 'lastSaturday'

LAST_SEVEN_DAYS = 'last7Days'

LAST_SUNDAY = 'lastSunday'

LAST_THREE_DAYS = 'last3Days'

LAST_THREE_HOURS = 'last3Hours'

LAST_THURSDAY = 'lastThursday'

LAST_TUESDAY = 'lastTuesday'

LAST_TWENTY_FOUR_HOURS = 'last24Hours'

LAST_TWO_DAYS = 'last2Days'

LAST_TWO_HOURS = 'last2Hours'

LAST_WEDNESDAY = 'lastWednesday'

LAST_WEEK = 'lastWeek'

THIS_MONTH = 'thisMonth'

THIS_WEEK = 'thisWeek'

TODAY = 'today'

class pydolphinscheduler.tasks.dependent.DependentItem(project_name: str, process_definition_name: str, dependent_task_name: str | None = '0', dependent_date: DependentDate | None = 'today')

Bases: Base

Dependent item object, minimal unit for task dependent.

It declare which project, process_definition, task are dependent to this task.

get_code_from_gateway() → Dict

Get project, definition, task code from given parameter.

_DEFINE_ATTR: set = {'cycle', 'date_value', 'definition_code', 'dep_task_code', 'project_code'}

property code_parameter: Tuple

Get name info parameter to query code.

property cycle: str

Get dependent cycle.

property date_value: str

Get dependent date.

property definition_code: str

Get dependent definition code.

property dep_task_code: str

Get dependent tasks code list.

property is_all_task: bool

Check whether dependent all tasks or not.

property project_code: str

Get dependent project code.

class pydolphinscheduler.tasks.dependent.DependentOperator(*args)

Bases: Base

Set DependentItem or dependItemList with specific operator.

get_define(camel_attr=True) → Dict

Overwrite Base.get_define to get task dependent specific get define.

classmethod operator_name() → str

Get operator name in different class.

set_define_attr() → str

Set attribute to function get_define().

It is a wrapper for both And and Or operator.

_DEFINE_ATTR: set = {'relation'}

property relation: str

Get operator name in different class, for function get_define().

class pydolphinscheduler.tasks.dependent.Or(*args)

Bases: DependentOperator

Operator Or for task dependent.

It could accept both DependentItem and children of DependentOperator, and set OR condition to those args.