CODEWITHSUNDEEP
pythonairflow
Arya
Arya

Reputation: 531

KeyError: 'Variable template_fields does not exist'

while i am using get_current_context() and executing the python operator task as below getting error: variable template field doesnt exist.

@task
def varfile(regularvalue,previousvalue,dag_instance, **kwargs):
    if regularvalue:
        context = get_current_context()
        varvalue = PythonOperator(
                        task_id=f"varvalue",
                        python_callable=def_varvalue,
                        provide_context= True,
                        op_kwargs = {"dag_name":dag_name,"regularvalue":regularvalue,"previousvalue":int(previousvalue)},
                        dag=dag_instance
                        )
        varvalue.execute(context)
        varvalue

finalval = varfile("{{ params.regvalue }}","{{ params.regvalue_previous_date }}", dag_instance=dag)


**Error:** KeyError: 'Variable template_fields does not exist'
  File "/home/dag.py", line 100, in varfile
    varvalue.execute(context)
  File "/home/airflow/operators/python.py", line 148, in execute
    self.op_kwargs = determine_kwargs(self.python_callable, self.op_args, context)
  File "/home/airflow/models/baseoperator.py", line 742, in __setattr__
    self.set_xcomargs_dependencies()
  File "/home/airflow/models/baseoperator.py", line 864, in set_xcomargs_dependencies
    apply_set_upstream(arg)
  File "/home/airflow/models/baseoperator.py", line 856, in apply_set_upstream
    apply_set_upstream(elem)
  File "/home/airflow/models/baseoperator.py", line 856, in apply_set_upstream
    apply_set_upstream(elem)
  File "/home/airflow/models/baseoperator.py", line 857, in apply_set_upstream
    elif hasattr(arg, "template_fields"):
  File "/home/airflow/models/taskinstance.py", line 1663, in __getattr__
    self.var = Variable.get(item, deserialize_json=True)
  File "/home/airflow/models/variable.py", line 140, in get
    raise KeyError(f'Variable {key} does not exist')
KeyError: 'Variable template_fields does not exist'

Did i miss something , any suggestion pls

Upvotes: 0

Views: 2873

Answers (1)

Hussein Awala
Hussein Awala

Reputation: 5096

You don't need to create a PythonOperator instance if you use the task decorator:

@task
def varfile(regularvalue,previousvalue,dag_instance, **kwargs):
    if regularvalue:
        context = get_current_context()
        varvalue = def_varvalue(dag_name=dag_name, regularvalue=regularvalue, previousvalue=int(previousvalue), dag=dag_instance) # maybe you don't need all these args
        return varvalue

finalval = varfile("{{ params.regvalue }}","{{ params.regvalue_previous_date }}", dag_instance=dag)

Upvotes: 1

Related Questions