add default args

from airflow import DAG

from airflow.operators.bash import BashOperator

from airflow.providers.standard.operators.bash import BashOperator

from datetime import datetime, timedelta

from utils import TRIGGER_DEFAULT_ARGS

default_args = {

    'owner': 'airflow',

    'depends_on_past': False,

    'email': ['bozhang@nao.cas.cn'],

    'email_on_failure': False,

    'email_on_retry': False,

    'retries': 1,

    'retry_delay': timedelta(minutes=5),

    'params': {

        'example_param': 'example_value',

        'a':1,

        'b':2,

        'c':3

    },

}

# 添加默认参数

default_args = dict(

    params=dict(

        dataset="csst-msc-c9-25sqdeg-v3",

        instrument="MSC",

        obs_type="WIDE",

        obs_group="W2",

        obs_id="10100232366",

        detector="09",

        pmapname="",

        ref_cat="",

        batch_id="airflow-batch",

        priority=1,

    ),

    **TRIGGER_DEFAULT_ARGS,

)

with DAG(

    dag_id='csst-msc-l1.trigger',

    dag_id="csst-msc-l1.trigger",

    default_args=default_args,

    description='A simple tutorial DAG',

    start_date=datetime(2023, 1, 1),

    description="CSST MSC L1 pipeline trigger.",

    start_date=datetime(2025, 1, 1),

    schedule="@daily",

    catchup=False,

    tags={'MSC', 'Trigger'},

    tags={"MSC", "L1", "Trigger"},

) as dag:

    t1 = BashOperator(

        task_id='print_date',

        bash_command='date',

        task_id="print_date",

        bash_command="date",

    )

    t2 = BashOperator(

        task_id='sleep',

        bash_command='sleep 5',

        task_id="sleep",

        bash_command="sleep 5",

        retries=3,

    )

    t1 >> t2

from ._args import TRIGGER_DEFAULT_ARGS

from ._utils import get_run_id_task

from datetime import timedelta

TRIGGER_DEFAULT_ARGS = {

    "owner": "airflow",

    "depends_on_past": False,

    "retries": 1,

    "retry_delay": timedelta(minutes=5),

    "email": ["bozhang@nao.cas.cn"],

    "email_on_failure": False,

    "email_on_retry": False,

    # "params": {},

    "trigger_rule": "all_success",

}

from airflow.providers.standard.operators.python import PythonOperator

def get_run_id(**context):

    # 从上下文中直接获取 dag_run_id

    current_run_id = context["dag_run"].run_id

    print(f"Current run_id: {current_run_id}")

get_run_id_task = PythonOperator(

    task_id="get_dag_run_id",

    python_callable=get_run_id,

    provide_context=True,  # Airflow 2.x 默认启用，可省略

)