TypeError: PostgresOperator.partial() got an unexpected keyword argument 'schedule_interval'

Vlad Vlad 0
Viewed 29

So I am trying to use partial() method in PostgresOperator, but I am getting this error apparently because I unintentionally pass schedule_interval to it. I looked up in airflow git there's no such parameter for partial() method of BasicOperator which I assume is parent for all the operators classes.

So I am confused I have to pass this parameter to DAG params yet there's no such parameter for .partial() so how am I supposed to create this dag and tasks? I haven't found any information on how to pull it off.

from airflow import DAG
from airflow.decorators import task
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2022, 9, 7),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'schedule_interval' : '@daily'
}


with DAG(
    'name',
    default_args=default_args    
) as dag:


    @task
    def generate_sql_queries(src_list: list) -> list:
        queries = []
        for i in src_list:
            query = f'SELECT sql_epic_function()'
            queries.append(query)
        return queries

    queries = generate_sql_queries([4,8])



    task = PostgresOperator.partial(
        task_id='name',
        postgres_conn_id='postgres_default_id_connection' # don't forget to change

    ).expand(sql=queries)


    task
airflow
1 Answers

The values in default_args dict are passed to each Operator, and you have 'schedule_interval' : '@daily' on it. Also, schedule_interval is not an operator argument but a DAG object argument. So, besides removing it from the default_args dict, you have to add it to the DAG definition like:

with DAG(
    'name',
    schedule_interval="@daily"
    default_args=default_args    
) as dag:
Javier Lopez Tomas0
Related