Airflow conf parameter in PostgresHook throwing error

user_3858193 0
Viewed 1052

The script takes input params from airflow conf {"flag":"NA","metric_name":"RED"}. I am trying to use the one param value in where clause of a sql. Can some one check if any issues the way I am passing the where caluse?

ERROR: - syntax error at or near "{"

def get_metrics(**kwargs):
    varteam_flag=(kwargs['dag_run'].conf['flag'])
    print("flag :",varteam_flag)
    params={"param1": varteam_flag}
    print(" params :",params['param1'])
    conn_id = kwargs.get('conn_id')
    pg_hook = PostgresHook(conn_id)
    sql="select count(1) as cnt FROM odw.metrics where  flag = {{ params.param1 }} "
    print(sql)
    records = pg_hook.get_records(sql)
    print("Records count is  :",str(records))
    return records

getData_metrics=PythonOperator(task_id='getData_metrics', python_callable=get_metrics,op_kwargs {'conn_id':'veas'},provide_context=True, dag=dag)
airflow
2 Answers

You can't use Jinja inside a python callable the way you do. Jinja is being evaluated when the operator is called, so in your case the jinja engine does not apply on the sql statement.

What you need to do is just use a simple assignment as:

sql=f"select count(1) as cnt FROM odw.metrics where  flag = {params.param1} "

I'm not so sure why you do so many assignments I believe

sql=f"select count(1) as cnt FROM odw.metrics where flag = {varteam_flag} "

should provide the same sql statement.

user_146244090

How to pass command line parameter/conf parameter in UI and use it in PostgresHook/PostgresOperator [simple python variable works here either using f or .format()].

1. If you are using PostgresHook and calling PythonOperator for the module
def get_metrics(**kwargs):
    varteam_flag=(kwargs['dag_run'].conf['flag'])
    conn_id = kwargs.get('conn_id')
    pg_hook = PostgresHook(conn_id)
    connection = pg_hook.get_conn()
    sql=f"select count(1) as cnt FROM odw.metrics where  flag = {varteam_flag} "
    print(sql)
    cursor = connection.cursor()
    cursor.execute(sql)
    rows = cursor.fetchall()
    cols = list(map(lambda x: x[0], cursor.description))

2. if you are using PostgresOperator [you can use jinja template]

gerics=PostgresOperator(task_id='gerics',
                     postgres_conn_id='veritas',
                      params={'param1': 10 },
                     sql=['truncate table test.dept1','insert into test.dept1 (select * from test.dept where rating={{ params.param1 }} )'],

Let me know if any other way to write.

user_38581930
Related