Airflow API payload as input to python script

Question

I am triggering a Airflow DAG using the following python script: API Call:

import requests
import json

url = "http://127.0.0.1:8080/api/v1/dags/get_external_params/dagRuns"

payload = json.dumps({
  "conf": {
    "ip": "198.51.100.96",
    "hostname": "abc",
    "domainname": "server456.com",
    "username": "airflow",
    "password": "XXXX"
  }
})
headers = {
  'Content-Type': 'application/json',
  'Authorization': 'Basic XXXX'
}

response = requests.request("POST", url, headers=headers, data=payload)

print(response.text)

I am trying to print the conf values all at once instead of printing them one by one. All I can see in examples is by having one key-value pair message and printing them DAG

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

def run_this_func(**kwargs):
    dag_run = kwargs.get('dag_run')
    parameters = dag_run.conf['ip'] #get all parameters instead of one key
    return parameters

args = {
    'owner': 'airflow',
}

with DAG(
    dag_id='get_external_params',
    default_args=args,
    schedule_interval=None,
    start_date=days_ago(2),
    tags=['DNS'],
) as dag:


   run_this = PythonOperator(
     task_id='run_this',
     python_callable=run_this_func,
     dag=dag,
     #op_kwargs = How can I get all the ip,hostname,username,password here
     provide_context=True,
   )
   runthis

my intention is to call a python script with arguments ex: somename.py -username xxxx -password xxx but before that I am trying to get the values inside the Airflow script. Please help!!!

Answer 1

I'm not sure if I understand your question 100% correctly, but dag_run.conf is a dict ({'ip': '198.51.100.96', 'hostname': 'abc', 'domainname': 'server456.com', 'username': '***', 'password': 'XXXX'}), so you can extract all keys you like from it. There is no way to assign keys/values to variables shorter though:

parameters = dag_run.conf
ip = parameters["ip"]
hostname = parameters["hostname"]
domainname = parameters["domainname"]

You can print the complete object. The full DAG:

from airflow.models import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

dag = DAG(dag_id="demo", start_date=days_ago(3), schedule_interval=None)


def _do_magic(**context):
    print(context["dag_run"].conf)
    # Will print "{'ip': '198.51.100.96', 'hostname': 'abc', 'domainname': 'server456.com', 'username': '***', 'password': 'XXXX'}"


do_magic = PythonOperator(task_id="do_magic", python_callable=_do_magic, dag=dag)

Since op_kwargs is template-able, you can also provide these as Jinja-templated strings:

from airflow.models import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

dag = DAG(dag_id="demo", start_date=days_ago(3), schedule_interval=None)


def _do_magic(ip):
    print(ip)
    # Will print 198.51.100.96


do_magic = PythonOperator(
    task_id="do_magic",
    python_callable=_do_magic,
    op_kwargs={"ip": "{{ dag_run.conf['ip'] }}"},
    dag=dag,
)

Does this answer?

Answer 2

IMHO The easiest way to achieve that is that is using params instead of op_kwargs. Consider this example request:

import requests
import json

url = "localhost:8080/api/v1/dags/api_triggered_with_params/dagRuns"

payload = json.dumps({
  "conf": {
    "param_via_API": "triggered from stable API"
  }
})
headers = {
  'Content-Type': 'application/json',
  'Authorization': 'Basic xxxx'
}

response = requests.request("POST", url, headers=headers, data=payload)

print(response.text)

DAG:

from airflow import DAG
from airflow.models.baseoperator import chain
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago


def _print_params(**kwargs):
    print(f"Task_id: {kwargs['ti'].task_id}")
    for k, v in kwargs["params"].items():
        print(f"{k}:{v}")

dag = DAG(
    dag_id="api_triggered_with_params",
    default_args={"owner": "airflow"},
    start_date=days_ago(1),
    schedule_interval="@once",
    tags=["example_dags"],
    params={"param_at_dag_level": "param_66"},
    catchup=False,
)
with dag:

    python_task = PythonOperator(
        task_id="python_task",
        python_callable=_print_params,
    )
    python_task_2 = PythonOperator(
        task_id="python_task_2",
        python_callable=_print_params,
        params={"param4": "param defined at task level"},
    )

chain(python_task, python_task_2)

Within the python callable, you can access any params defined at any level (DAG level, task or defined in conf when triggering the DagRun).

Output logs:

python_task

[2021-07-28 13:17:24,716] {logging_mixin.py:104} INFO - Task_id: python_task
[2021-07-28 13:17:24,717] {logging_mixin.py:104} INFO - param_at_dag_level:param_66
[2021-07-28 13:17:24,717] {logging_mixin.py:104} INFO - param_via_API:triggered from stable API
[2021-07-28 13:17:24,717] {python.py:151} INFO - Done. Returned value was: None

python_task_2

[2021-07-28 13:17:26,034] {logging_mixin.py:104} INFO - Task_id: python_task_2
[2021-07-28 13:17:26,034] {logging_mixin.py:104} INFO - param_at_dag_level:param_66
[2021-07-28 13:17:26,035] {logging_mixin.py:104} INFO - param4:param defined at task level
[2021-07-28 13:17:26,035] {logging_mixin.py:104} INFO - param_via_API:triggered from stable API
[2021-07-28 13:17:26,035] {python.py:151} INFO - Done. Returned value was: None

Let me know if that worked for you!