AWS Airflow trigger with configuration JSON

i try to make dynamic trigger with airflow, means during trigger airflow i want to change multiple parameter before i trigger the scripts.. right now i stuck on how to pass the parameter using configuration json UI provide by aws airflow.

airflow scripts as below :

    from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime

def print_hello():
    message = 'hello world'
    print(message)

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 8, 7),
    'retries': 1,
}

dag = DAG(
    'hello_world_dag',
    default_args=default_args,
    schedule_interval=None,
    catchup=False,
)

print_hello_task = PythonOperator(
    task_id='print_hello_task',
    python_callable=print_hello,
    dag=dag,
)

print_hello_task

this example i pass the parameter into configuration json.

how i can pass parameter "message" from configuration json to the scripts?

Another option in to use Jinja Template {{ dag_run.conf }} as mention in the ui

in this case very important to add render_template_as_native_obj=True to the Dag definition cause you want to consider this conf as dict.

the conf would be passed to the function as kwargs

with DAG(
    dag_id="hello_world_dag",
    schedule_interval=None,
    default_args=default_args,
    render_template_as_native_obj=True,
    catchup=False,) as dag:

    def print_hello(**kwargs):
        print(kwargs.get("message"))


    print_hello_task = PythonOperator(
        task_id='print_hello_task',
        python_callable=print_hello,
        op_kwargs="{{ dag_run.conf }}",
    )

    print_hello_task

Here's an example DAG that prints some properties of a provided dag run config object of {"hello": "world"} using Airflow's TaskFlow API (see the context section for more info on accessing a task instance's context):

from pendulum import datetime

from airflow.decorators import (
    dag,
    task,
)


@dag(
    schedule="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False,
    default_args={
        "retries": 2,
    },
    tags=["example"],
)
def read_config():
    @task
    def read_conf(dag_run=None):
        print(type(dag_run.conf))
        print(f"DAG Run config: {dag_run.conf}")
        print(f"Hello {dag_run.conf['hello']}")

    read_conf()


read_config()

Task instance logs:

[2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - <class 'dict'>
[2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - DAG Run config: {'hello': 'world'}
[2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - Hello world