Airflow Variables & Jinja

{
    'ds': '2026-05-08',                    # Execution date as a string (Dynamic)
    'ds_nodash': '20260508',               # Date string without dashes
    'dag': <DAG: my_pipeline>,             # The current DAG object
    'task': <Task: my_task>,               # The current Task object
    'run_id': 'manual__2026-05-08...',     # Unique ID for this specific run 
    'params': {},                          # User-defined runtime parameters
    'logical_date': datetime(...),         # Precise timestamp of the run
    'data_interval_start': datetime(...),  # Start of the data window
    'data_interval_end': datetime(...),    # End of the data window
    'task_instance': <TaskInstance: ...>,  # Metadata about this task attempt
    # ... and many other metadata keys
}

# ===================== Import Libraries =====================
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime


def print_jinja_report(dag_id, task_id, run_date, run_id):
    print("====== Pipeline Report ======")
    print(f" DAG ID   : {dag_id}")
    print(f" Task ID  : {task_id}")
    print(f" Run Date : {run_date}")
    print(f" Run ID   : {run_id}")
    print("  Status   : Success")
    print("=============================")

# ===================== Report Function =====================
def print_context_report(**context):
    print("====== Pipeline Report ======")
    print(f" DAG ID   : {context['dag'].dag_id}")
    print(f" Task ID  : {context['task_instance'].task_id}")
    print(f" Run Date : {context['ds']}")
    print(f" Run ID   : {context['run_id']}")
    print("  Status   : Success")
    print("=============================")

# ===================== DAG Definition =====================
with DAG(
    dag_id='context_report_dag',
    start_date=datetime(2026, 4, 20),
    schedule_interval=None,
    catchup=False
) as dag:

    # ===================== Generate Context Report Task =====================
    generate_context_report = PythonOperator(
        task_id='generate_context_report',
        python_callable=print_context_report
    )

    # ===================== Generate Context jinja Task =====================
    generate_python_jinja_report = PythonOperator(
        task_id='generate_python_jinja_report',
        python_callable=print_jinja_report,
        op_kwargs={
            'dag_id': '{{ dag.dag_id }}',
            'task_id': '{{ task.task_id }}',
            'run_date': '{{ ds }}',
            'run_id': '{{ run_id }}'
        }
    )

    # ===================== Generate Jinja Report (Bash) =====================
    generate_bash_jinja_report = BashOperator(
        task_id='generate_bash_jinja_report',
        bash_command="""
        echo "====== Pipeline Report ======"
        echo " DAG ID   : {{ dag.dag_id }}"
        echo " Task ID  : {{ task.task_id }}"
        echo " Run Date : {{ ds }}"
        echo " Run ID   : {{ run_id }}"
        echo " Status   : Success"
        echo "============================="
        """
    )

    generate_context_report >> generate_python_jinja_report >> generate_bash_jinja_report

Admin → Variables → + Add

docker compose exec airflow-webserver bash

airflow variables set start_date 1990-01-01
airflow variables set end_date   2000-12-31

from airflow.models import Variable

# ── In Python ─────────────────────────────────────────────────
start = Variable.get("start_date")
end   = Variable.get("end_date")

-- In Jinja (inside SQL or bash_command )

{{ var.value.start_date }}
{{ var.value.end_date }}

# ===================== Import Libraries =====================
from airflow import DAG
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime
from airflow.models import Variable

# ===================== Default Arguments =====================
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2026, 4, 20),
}

# ===================== DAG Definition =====================
with DAG(
    dag_id='postgres_operator_dag',
	catchup=False,
    schedule_interval=None,
    default_args=default_args
    
) as postgres_operator_dag:
	
	# ===================== Airflow Variables =====================
    start = Variable.get("start_date") 
    end = Variable.get("end_date")

    # ===================== Create Table =====================
    create_table = PostgresOperator(
        task_id='create_table',
        postgres_conn_id='postgres_conn',
        sql="""
            CREATE TABLE IF NOT EXISTS customers (
                customer_id   VARCHAR(50) PRIMARY KEY,
                customer_name VARCHAR NOT NULL,
                address       VARCHAR NOT NULL,
                birth_date    DATE NOT NULL
            );
        """
    )

    # ===================== Insert Data =====================
    insert_values = PostgresOperator(
        task_id='insert_values',
        postgres_conn_id='postgres_conn',
        sql='./scripts/insert.sql'
    )

    # ===================== Select Data =====================
    select_values = PostgresOperator(
        task_id='select_values',
        postgres_conn_id='postgres_conn',
        sql="""
            SELECT *
            FROM customers
            WHERE birth_date BETWEEN %(start_date)s AND %(end_date)s;
        """,
        parameters={
	        'start_date': start,  # 💡 Exercise: Try alternative '{{ var.value.start_date }}'
	        'end_date':   end     # 💡 Exercise: Try alternative '{{ var.value.end_date }}'
	    }
    )

    # ===================== Task Dependencies =====================
    create_table >> insert_values >> select_values

# ===================== Import Libraries =====================
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime

# ===================== Python Function =====================
def python_task(**context):
    print("====== Python Task ======")
    print(f"User Name : {context['params']['user_name']}")
    print(f"File Name : {context['params']['file_name']}")
    print(f"Separator  : {context['params']['separator']}")
    print("=========================")

# ===================== DAG =====================
with DAG(
    dag_id='params_mixed_example',
    start_date=datetime(2026, 4, 20),
    schedule_interval=None,
    catchup=False,
    params={
        "user_name": "Diea",
        "file_name": "sales.csv"
    }
) as dag:

    task_python = PythonOperator(
        task_id='python_task',
        python_callable=python_task,
        params={"separator": ","}
    )

    task_bash = BashOperator(
        task_id='bash_task',
        bash_command="""
        echo "====== Bash Task ======"
        echo "User Name: {{ params.user_name }}"
        echo "File Name: {{ params.file_name }}"
        echo "Separator : {{ params.separator }}"
        echo "======================="
        """,
        params={"separator": ";"}
    )

    task_python >> task_bash