Airflow: Running PostgresOperator within a PythonOperator with Taskflow API

The goal I am having is to loop over a list that I get from another PythonOperator and within this loop save the json to the Postgres DB. I am using the Taskflow API from Airflow 2.0.

The code works well if I write the SQL statement directly to the sql parameter in the PostgresOperator. But when I write the SQL to a file and put the file path into the sql parameter, then the error is thrown:

psycopg2.errors.SyntaxError: syntax error at or near "sql"
LINE 1: sql/insert_deal_into_deals_table.sql

This is the code of the task:

    @task()
    def write_all_deals_to_db(all_deals):
        for deal in all_deals:
            deal_json = json.dumps(deal)
            pg = PostgresOperator(
                task_id='insert_deal',
                postgres_conn_id='my_db',
                sql='sql/insert_deal_into_deals_table.sql',
                params={'deal_json': deal_json}
            )
            pg.execute(dict())

The weird thing is, that the code works if I use it as a standalone Operator (outside of a PythonOperator). Like this:

create_deals_table = PostgresOperator(
    task_id='create_deals_table',
    postgres_conn_id='my_db',
    sql='sql/create_deals_table.sql'
)

I tried around a lot and I guess that it has to do with the Jinja templating. Somehow within a PythonOperator the PostgresOperator cannot make use of neither the param nor the .sql file parsing.

Any tip or reference is greatly appreciated!

EDIT:

This code works, but is rather a quick fix. The actual problem I am still having, is that the Jinja template is not working for the PostgresOperator when I am using it inside a PythonOperator.

@task()
def write_all_deals_to_db(all_deals):
    sql_path = 'sql/insert_deal_into_deals_table.sql'
    for deal in all_deals:
        deal_json = _transform_json(deal)
        sql_query = open(path.join(ROOT_DIRECTORY, sql_path)).read()
        sql_query = sql_query.format(deal_json)
        pg = PostgresOperator(
            task_id='insert_deal',
            postgres_conn_id='my_db',
            sql=sql_query
        )
        pg.execute(dict())