how to pass parameters from pythonoperator task to simplehttpoperator task in airflow dag?

Question

i want to trigger a simplehttpoperator,like this: airflow trigger_dag test_trigger --conf '{"name":"something"}'

then i use a pythonoperator python_callable to accept parameters by using kwargs['dag_run'].conf , and i want to pass the ['dag_run'].conf to simplehttpoperator, how can i do it? anyone can help?

cc_ = {}


def run_this_func(ds, **kwargs):
    cc_ = kwargs['dag_run'].conf
    logging.info(cc_)
    return cc_

run_this = PythonOperator(
    task_id='run_this',
    provide_context=True,
    python_callable=run_this_func,
    dag=dag)

http_task = SimpleHttpOperator(
    task_id='http_task',
    http_conn_id='test_http',
    method='POST',
    endpoint='/api/v1/function',
    data=cc_,
    headers={"Authorization": "Basic YWRtaW46MTIzNDU2", "Accept": "application/json, text/plain, */*"},
    response_check=lambda response: True if "10000" in response.content else False,
    dag=dag)

http_task.set_upstream(run_this)

score 2 · Answer 1 · answered Sep 14 '17 at 07:11

Thanks to @Chengzhi and @Daniel. At last I wrote a custom filter 'tojson' in Jinja2/filter.py,because in airflow the default Jinja2 version is 2.8.1 and Jinja2 does not contain the builtin filter named 'tojson' until version 2.9 .

def do_tojson(value):
    value = json.JSONEncoder().encode(value)
    return value

In dag file, the code as follows. It works.

def run_this_func(ds, **kwargs):
    cc_ = kwargs['dag_run'].conf
    return cc_

run_this = PythonOperator(
    task_id='run_this',
    provide_context=True,
    python_callable=run_this_func,
    dag=dag)

http_task = SimpleHttpOperator(
    task_id='http_task',
    http_conn_id='test_http',
    method='POST',
    endpoint='/api/v1/task',
    data="{{ task_instance.xcom_pull(task_ids='run_this') |tojson}}",
    headers={"Authorization": "Basic YWRtaW46MTIzNDU2", "Accept": "application/json, text/plain, */*",
             "Content-Type": "application/json"},
    response_check=lambda response: True if "10000" in response.content else False,
    dag=dag)

http_task.set_upstream(run_this)

Chengzhi · Answer 2 · 2017-09-12T19:12:21.750

0

For the communication between tasks, you might want to check the XCOM, https://airflow.incubator.apache.org/concepts.html#xcoms

*****UPDATE*****
(thanks Daniel for more detail) Below is some codes you can give it a try, in your SimpleHttpOperator you get the return value via XCOM:

http_task = SimpleHttpOperator(
    task_id='http_task',
    http_conn_id='test_http',
    method='POST',
    endpoint='/api/v1/function',
    data=json.loads("{{ task_instance.xcom_pull(task_ids='run_this', key='return_value') }}"),
    headers={"Authorization": "Basic YWRtaW46MTIzNDU2", "Accept": "application/json, text/plain, */*"},
    response_check=lambda response: True if "10000" in response.content else False,
    dag=dag)

edited Sep 12 '17 at 19:12

answered Sep 11 '17 at 15:40

Chengzhi

2,531
2
27
41

but how to use XCOM in simplehttpoperator? can you give any case code? – pyfroggogogo Sep 12 '17 at 08:41
@pyfroggogogo, I update with some code example, try if this works – Chengzhi Sep 12 '17 at 12:42
1

Templates have to be passed as a string. You can use `data=json.loads('{{ ... | tojson }}')` to get it back into a dict type after rendering. – Daniel Huang Sep 12 '17 at 17:31

score 0 · Answer 3 · answered Oct 04 '21 at 22:42

0

You dont need a separate PythonOperator to collect data. I created a custom operator which can accept data (passed from an API) and pass it to send a GET?POST request. Here is the link: https://stackoverflow.com/a/69443084/8081381

answered Oct 04 '21 at 22:42

Vinit Bodhwani

413
1
5
16

how to pass parameters from pythonoperator task to simplehttpoperator task in airflow dag?

3 Answers3

Linked