snowflake.core.task.dagv1.DAG

class snowflake.core.task.dagv1.DAG(name: str, *, schedule: Cron | timedelta | None = None, warehouse: str | None = None, user_task_managed_initial_warehouse_size: str | None = None, error_integration: str | None = None, comment: str | None = None, allow_overlapping_execution: bool | None = None, user_task_timeout_ms: int | None = None, suspend_task_after_num_failures: int | None = None, config: Dict[str, Any] | None = None, session_parameters: Dict[str, Any] | None = None, stage_location: str | None = None, imports: List[str | Tuple[str, str]] | None = None, packages: List[str | ModuleType] | None = None, use_func_return_value: bool = False)

Bases: object

A graph of tasks composed of a single root task and additional tasks, organized by their dependencies.

Snowflake doesn’t have a first-class task graph entity, so this is a client-side object representation which manages Task relationship. A root Task and its successors logically form a task graph or DAG (Directed Acyclic Graph). Refer to Task graphs.

When a task graph is deployed, all child tasks are created in Snowflake. A dummy Task is created as the root. A task’s predecessor is the dummy task if it’s added to the task graph with no other predecessors.

Example

>>> dag = DAG("TEST_DAG",
...     schedule=timedelta(minutes=10),
...     use_func_return_value=True,
...     warehouse="TESTWH_DAG",
...     packages=["snowflake-snowpark-python"],
...     stage_location="@TESTDB_DAG.TESTSCHEMA_DAG.TEST_STAGE_DAG"
... )
>>> def task1(session: Session) -> None:
...     session.sql("select 'task1'").collect()
>>> def task2(session: Session) -> None:
...     session.sql("select 'task2'").collect()
>>> def cond(session: Session) -> str:
...     return 'TASK1'
>>> with dag:
...     task1 = DAGTask("TASK1", definition=task1, warehouse="TESTWH_DAG")
...     task2 = DAGTask("TASK2", definition=task2, warehouse="TESTWH_DAG")
...     condition = DAGTaskBranch("COND", definition=cond, warehouse="TESTWH_DAG")
...     condition >> [task1, task2]
>>> dag_op = DAGOperation(schema)
>>> dag_op.deploy(dag, mode="orReplace")
>>> dag_op.run(dag)
Note:
    When defining a task branch handler, simply return the task name you want to jump to. The task name is
    case-sensitive, and it has to match the name property in DAGTask. For exmaple, in above sample code, return
    'TASK1' instead of 'TEST_DAG$TASK1', 'task1' or 'Task1' will not be considered as a exact match.
Copy

Refer to snowflake.core.task.Task for the details of each property.

Attributes

tasks

Returns a list of tasks this task graph has.

name

Name of the task graph and the dummy root task.

warehouse

Refer to snowflake.core.task.Task.warehouse.

user_task_managed_initial_warehouse_size

Refer to snowflake.core.task.Task.user_task_managed_initial_warehouse_size.

comment

comment of the task graph.

schedule

Schedule of the task graph. Refer to snowflake.core.task.Task.schedule.

error_integration

Refer to snowflake.core.task.Task.error_integration.

allow_overlapping_execution

Refer to snowflake.core.task.Task.allow_overlapping_execution.

user_task_timeout_ms

Refer to snowflake.core.task.Task.user_task_timeout_ms.

suspend_task_after_num_failures

Refer to snowflake.core.task.Task.suspend_task_after_num_failures.

config

Refer to snowflake.core.task.Task.config.

session_parameters

Refer to snowflake.core.task.Task.session_parameters.

stage_location

The default stage location where this task graph’s tasks code will be stored if creating the tasks from Python functions.

imports

The default imports for all tasks of this task graph if creating the tasks from Python functions.

packages

The default packages for the tasks of this task graph if creating the tasks from Python functions.

use_func_return_value

Use the Python function’s return value as Task return value if use_func_return_value is True. Default False.

Methods

__init__(name: str, *, schedule: Cron | timedelta | None = None, warehouse: str | None = None, user_task_managed_initial_warehouse_size: str | None = None, error_integration: str | None = None, comment: str | None = None, allow_overlapping_execution: bool | None = None, user_task_timeout_ms: int | None = None, suspend_task_after_num_failures: int | None = None, config: Dict[str, Any] | None = None, session_parameters: Dict[str, Any] | None = None, stage_location: str | None = None, imports: List[str | Tuple[str, str]] | None = None, packages: List[str | ModuleType] | None = None, use_func_return_value: bool = False) → None
add_task(task: DAGTask) → None

Add a child task to this task graph.

get_task(task_name: str) → DAGTask | None

Get a child task from this task graph based on task name.

Language: English