Airflow Study10-날짜 템플릿

    @task(task_id="python_task")
    def show_templates(**kwarges):
        from pprint import pprint 
        pprint(kwarges)
        
    show_templates()

일배치, 매일 0시 30분에 수행한다고 했을때
수행시점 : 2024년 6월 2일 06시

dags_python_template.py

from airflow import DAG
import datetime
import pendulum
from airflow.operators.python import PythonOperator
from airflow.decorators import task

with DAG(
    dag_id = 'dags_python_template',
    schedule = '30 9 * * * ',
    start_date = pendulum.datetime(2024,6,1, tz = 'UTC'),
    catchup=False
) as dag :
    
    def python_function1(start_date, end_date, **kwargs):
        print(start_date)
        print(end_date)
        
    python_t1 = PythonOperator(
        task_id = 'python_t1',
        python_callable=python_function1 , 
        op_kwargs={'start_date':'{{data_interval_start | ds}}','end_date':'{{data_interval_end | ds}}' }
    )
    
    
    @task(task_id = 'python_t2')
    def python_function2(**kwargs):
        print(kwargs)
        print('ds: '+kwargs['ds'])
        print('ts: ' + kwargs['ts'])
        print('data_interval_start: ' + str(kwargs['data_interval_start']))
        print('data_interval_end: ' + str(kwargs['data_interval_end']))
        print('task_instance: ' + str(kwargs['ti']))
        
        
        
        
        
    
    python_t1 >> python_function2()