CODEWITHSUNDEEP
airflow
Mohan
Mohan

Reputation: 151

How to read response body from SimpleHttpOperator

I am new to Airflow. I have written a code to submit HTTP Post using SimpleHttpOperator. In this case post request return a token, i need a help on how reading the response body.

get_templates = SimpleHttpOperator(
    task_id='get_templates',
    method='POST',
    endpoint='myendpoint',
    http_conn_id = 'myconnection',
    trigger_rule="all_done",
    headers={"Content-Type": "application/json"},
    xcom_push=True,
    dag=dag
)

Looks like POST was successful. Now my question is how to read the response body. This is the output of code, there is no errors [2019-05-06 20:08:40,518] {http_hook.py:128} INFO - Sending 'POST' to url: https://auth.reltio.com/oauth//token?username=perf_api_user&password=perf_api_user!&grant_type=password /usr/lib/python2.7/site-packages/urllib3/connectionpool.py:847: InsecureRequestWarning: Unverified HTTPS request is being made. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings InsecureRequestWarning)

Upvotes: 2

Views: 6460

Answers (2)

Galuoises
Galuoises

Reputation: 3283

If you use Airflow 2 xcom_push argument is not available in SimpleHttpOperator. In this case let's say you call /my-url in task call_api; to get the response and pass it to another task you need to read from the xcom return_value that is automatically defined by the SimpleHttpOperator as:

call_api = SimpleHttpOperator(
    task_id='call_api',
    http_conn_id=api_connection,
    method='GET',
    endpoint='/my-url',
    response_filter=lambda response: json.loads(response.text),
    log_response=True,  # Shows response in the task log
    dag=dag
)

def _read_response(ti):
    val = ti.xcom_pull(
        task_ids='call_api',
        key='return_value'
    )
    print(val)


read_response = PythonOperator(
    task_id='read_response',
    python_callable=_read_response,
    dag=dag
)

You can also specify dag_id in ti.xcom_pull to select the running dag.

Upvotes: 1

bosnjak
bosnjak

Reputation: 8624

The execute function of the SimpleHttpOperator returns the response.text (source). By looking at the Airflow documentation for XCom, you can see that:

... if a task returns a value (either from its Operator’s execute() method, or from a PythonOperator’s python_callable function), then an XCom containing that value is automatically pushed.

meaning the response body is pushed to the XCom and is available for downstream tasks to access.

For example, you could have a PythonOperator fetching it via:

response_body = context['task_instance'].xcom_pull(task_ids='get_templates')

Additionally, if you just want to log the response instead of process it, you can just set the log_response of the SimpleHttpOperator constructor to True.

Upvotes: 3

Related Questions