How can you override values in a Airflow DAG run using Configuration JSON

user_176661 0
Viewed 1746

Been hacking away at this concept but I can't seem to get it working.

I have multiple DAG's extracting information from a database. These are scheduled daily and I am using macros to fill in the correct Oracle date format for my date ranges. This works well if I want to rerun a certain instance as the correct date is used if I have to re-run something.

START = "{{ macros.ds_format(ds, '%Y-%m-%d', '%d-%b-%y') }}"
END = "{{ macros.ds_format(next_ds, '%Y-%m-%d', '%d-%b-%y') }}"
date_range = (f"{START}", f"{END}")

What I cannot seem to figure out is how to use the "Trigger JSON" prompt when I manually trigger a DAG to override those values.

I know that it is passed as a context and I should be able to grab the values but I have not found clear documentation.

Ultimately, I expect to retrigger a DAG and pass a JSON start and end date to override my deafault settings.

# for example if I want all of December 2020 in a file the JSON would be the following
{"START":"01-DEC-20", "END":"31-DEC-20"}

this is a sample DAG setup where I pass the values into a Python Operator

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago

START = "{{ macros.ds_format(ds, '%Y-%m-%d', '%d-%b-%y') }}"
END = "{{ macros.ds_format(next_ds, '%Y-%m-%d', '%d-%b-%y') }}"
date_range = (f"{START}", f"{END}")


def _print_context(**context):
    print(context)


with DAG(
    "example_parametrized_dag",
    schedule_interval=None,
    start_date=days_ago(2),
    tags=["example"],
) as dag:

    print_context = PythonOperator(
        task_id="print_context",
        python_callable=_print_context,
        provide_context=True,
        op_kwargs={
            "parameters": date_range,
        },
    )

In a perfect world I could check if I have override values but I do not know if it is possible?

airflow python
1 Answers

Well, it seems writing out the question made me realize what the answer was...

Given the following JSON,

{"startdate":"01-DEC-20","enddate":"31-DEC-20"}

I can change my variables to check for the existence of those values and set my results accordingly.

START = "{{ dag_run.conf['startdate'] if dag_run else macros.ds_format(ds, '%Y-%m-%d', '%d-%b-%y') }}"
END = "{{ dag_run.conf['enddate'] if dag_run else macros.ds_format(next_ds, '%Y-%m-%d', '%d-%b-%y') }}"

Irritatingly simple. and the results are reflected if you check the log data (and printed context).

user_1766610
Related