Airflow Python Operator - Get values from xcom pull for params

Question

I have a SQL file like below.

select * from table where data > {{ params.maxdt }}

And Im calling a function from python operator. So I need to pass maxdt value while calling that python operator.

class SQLTemplatedPythonOperator(PythonOperator):
    template_ext = ('.sql',)

table_list = [public.t1]

for table_name in table_list:
    pull = SQLTemplatedPythonOperator(
    task_id='export_{}'.format(table_name),dag=dag,
    templates_dict={'query': 'public.t1.sql'},
    params = {'table_name': table_name, 'maxdt': {{ti.xcom_pull(task_ids='push_result_{}')}}.format(table_name)},
    op_kwargs={'tablename':table_name},
    python_callable=pgexport,
    provide_context=True,
    )

There was a task called push_result_{}.format(table_name) do some process and push a data value to xcom. Now here I need to get that value and pass it inside my pull task. So my SQL templated query will get the value and pass it to the pgexport function.

pgexport - it will use postgrestogcs operator to push the results of the templated SQL query.

Unfortunately the syntax I used is not working. Can someone help me to fix this?

score 0 · Answer 1 · answered Oct 24 '20 at 14:57

0

It looks like you've several syntax issues.

The first one is the missing quotation marks around the string, so {{ti.xcom_pull(task_ids='push_result_{}')}}.format(table_name) should become "{{ti.xcom_pull(task_ids='push_result_{}')}}".format(table_name).

Now, because you're using the python format function, you've to escape the curly brackets as they're used by the formatter, so you need to change it to:

"{{{{ ti.xcom_pull(task_ids='push_result_{}') }}}}".format(table_name)

As you can see, curly brackets are escaping themselves, the result after the formatting for table "TABLENAME" would be the following string: {{ ti.xcom_pull(task_ids='push_result_TABLENAME') }} which is now an airflow macro that can be replaced on execution time.

FYI - you can also decide to use the old python formatting to avoid the brackets escaping, something like: "{{ ti.xcom_pull(task_ids='push_result_%s') }}" % "TABLENAME".

answered Oct 24 '20 at 14:57

Zacharya Haitin

338
2
9

I tried this, but its not working. Its generating the SQL syntax as `select * from public.m1 where timestamp > {{ ti.xcom_pull(task_ids='push_result_table_name') }}` – TheDataGuy Oct 24 '20 at 15:12
And I found this link(https://stackoverflow.com/a/46061162/6885516), seems its not supported. But not sure – TheDataGuy Oct 24 '20 at 15:18
That means `params` isn't a templated field, you should add `template_fields = ["params"]` to your operator. – Zacharya Haitin Oct 24 '20 at 15:19
My bad, I get what you're trying to achieve now. Using templates_dict is the right solution here. – Zacharya Haitin Oct 24 '20 at 15:26
templates_dict? - pass the values to the function, then I can render the template on the function? – TheDataGuy Oct 24 '20 at 15:31
Yes, but it's something you need to do on the Operator class level, so Airflow will know to run Jinja on those fields. – Zacharya Haitin Mar 27 '21 at 12:06

Airflow Python Operator - Get values from xcom pull for params

1 Answers1