【SnowflakeのTask/DAGについて)　Snowflake における Task / DAG の Python API 管理の概要について


## 特徴（Feature Status）
機能ステータス：一般提供（GA）
利用不可リージョン：政府クラウド（Government Regions）

## 概要：Snowflake タスクを Python で管理する
Snowflake では Python を使って以下のようなタスク処理が可能です：

SQL ステートメントの自動実行

ストアドプロシージャ（SP）の実行

Snowflake Scripting による制御ロジックの実行

さらに、複数タスクの依存関係を持った「タスクグラフ（DAG）」も定義できます

🧩 Task と TaskResource の違い
クラス名	役割
Task	タスクの定義情報（スケジュール・中身・依存など）
TaskResource	既存タスクに対する操作（実行・一時停止・削除など）

## 🔌 接続と初期化
```python
from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)
```
Root オブジェクトを使って、Snowflake の Python API 群にアクセスできるようになります。


## 🏗️ Task を作成する
下記でSQL実行タスクの作成を実施する。
Task()でSnowflake に登録するタスクの情報を Python コード上で表現するための定義を実施している。

```python
from datetime import timedelta
from snowflake.core.task import Task

my_task = Task(name="my_task", definition="SELECT 1;", schedule=timedelta(hours=1))
tasks = root.databases['my_db'].schemas['my_schema'].tasks
tasks.create(my_task)
```
definition: 実行する SQL（または後述の SP 呼び出し）
schedule: timedelta もしくは Cron で設定可能

StoredProcedure を呼び出すタスクの作成
```python
from snowflake.core.task import StoredProcedureCall

my_task2 = Task(
    "my_task2",
    StoredProcedureCall(dosomething, stage_location="@mystage"),
    warehouse="test_warehouse"
)
tasks.create(my_task2)
```
※ warehouse の指定が必須です。

## 🔁 タスクの更新（create_or_alter）

my_task = tasks["my_task"].fetch()
my_task.definition = "SELECT 2;"
my_task.schedule = timedelta(hours=2)
tasks["my_task"].create_or_alter(my_task)


## 📃 タスク一覧取得
task_iter = tasks.iter(like="my%")
for task in task_iter:
    print(task.name)

▶️ タスク操作：実行・一時停止・再開・削除
task_res = tasks['my_task']
task_res.execute()
task_res.suspend()
task_res.resume()
task_res.drop()

## 🧠 DAG（タスクグラフ）を構築する
DAG 構造の定義とデプロイを下記で実行できる。
DAGでDAGの名前の定義、スケジュールを定義する。その後にDAGTask()において各Taskの定義を行い、依存関係も定義する(`dag_task1 >> dag_task2`)。
```python
from snowflake.core.task.dagv1 import DAG, DAGTask, DAGOperation
from snowflake.core.task import StoredProcedureCall

with DAG("my_dag", schedule=timedelta(days=1)) as dag:
    dag_task1 = DAGTask("dagtask1", "MERGE INTO ...")
    dag_task2 = DAGTask(
        StoredProcedureCall(dosomething, stage_location="@mystage", packages=["snowflake-snowpark-python"]),
        warehouse="test_warehouse"
    )
dag_task1 >> dag_task2  # 依存関係（task1 → task2）

schema = root.databases["my_db"].schemas["my_schema"]
dag_op = DAGOperation(schema)
dag_op.deploy(dag)
```

## ⏰ Cron スケジュール + ブランチ付き DAG + 戻り値利用
スケジュールのオプションを指定した
```python
from snowflake.core.task.dagv1 import DAGTaskBranch

def task_branch_handler(session) -> str:
    return "task3"

with DAG(
    "my_dag",
    schedule=Cron("10 * * * *", "America/Los_Angeles"),
    stage_location="@mystage",
    use_func_return_value=True
) as dag:
    task1 = DAGTask("task1", task_handler, warehouse=test_warehouse)
    task1_branch = DAGTaskBranch("task1_branch", task_branch_handler, warehouse=test_warehouse)
    task2 = DAGTask("task2", task_handler, warehouse=test_warehouse)
    task3 = DAGTask("task3", task_handler, warehouse=test_warehouse, condition="1=1")

    task1 >> task1_branch
    task1_branch >> [task2, task3]
```
## 前後タスク間で値を受け渡す（TaskContext）

```python
from snowflake.core.task.context import TaskContext

def producer(session):
    ctx = TaskContext(session)
    ctx.set_return_value("result_123")

def consumer(session):
    ctx = TaskContext(session)
    value = ctx.get_predecessor_return_value("producer")
```