Stacktrace generated on Airflow's Webserver when I try to manually trigger DAG via WebUI

airflow_webserver_stacktrace.log

                          ____/ (  (    )   )  \___
                         /( (  (  )   _    ))  )   )\
                       ((     (   )(    )  )   (   )  )
                     ((/  ( _(   )   (   _) ) (  () )  )
                    ( (  ( (_)   ((    (   )  .((_ ) .  )_
                   ( (  )    (      (  )    )   ) . ) (   )
                  (  (   (  (   ) (  _  ( _) ).  ) . ) ) ( )
                  ( (  (   ) (  )   (  ))     ) _)(   )  )  )
                 ( (  ( \ ) (    (_  ( ) ( )  )   ) )  )) ( )
                  (  (   (  (   (_ ( ) ( _    )  ) (  )  )   )
                 ( (  ( (  (  )     (_  )  ) )  _)   ) _( ( )
                  ((  (   )(    (     _    )   _) _(_ (  (_ )
                   (_((__(_(__(( ( ( |  ) ) ) )_))__))_)___)
                   ((__)        \\||lll|l||///          \_))
                            (   /(/ (  )  ) )\   )
                          (    ( ( ( | | ) ) )\   )
                           (   /(| / ( )) ) ) )) )
                         (     ( ((((_(|)_)))))     )
                          (      ||\(|(|)|/||     )
                        (        |(||(||)||||        )
                          (     //|/l|||)|\\ \     )
                        (/ / //  /|//||||\\  \ \  \ _)
-------------------------------------------------------------------------------
Node: ip-XXX-XX-XX-XX.ap-southeast-1.compute.internal
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask/app.py", line 2292, in wsgi_app
    response = self.full_dispatch_request()
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask/app.py", line 1815, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask/app.py", line 1718, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask/_compat.py", line 35, in reraise
    raise value
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask/app.py", line 1813, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask/app.py", line 1799, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask_admin/base.py", line 69, in inner
    return self._run_view(f, *args, **kwargs)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask_admin/base.py", line 368, in _run_view
    return fn(self, *args, **kwargs)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/flask_login/utils.py", line 258, in decorated_view
    return func(*args, **kwargs)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/www/utils.py", line 369, in view_func
    return f(*args, **kwargs)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/www/utils.py", line 275, in wrapper
    return f(*args, **kwargs)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/utils/db.py", line 73, in wrapper
    return func(*args, **kwargs)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/www/views.py", line 1522, in tree
    'children': [recurse_nodes(t, set()) for t in dag.roots],
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/www/views.py", line 1522, in <listcomp>
    'children': [recurse_nodes(t, set()) for t in dag.roots],
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/www/views.py", line 1482, in recurse_nodes
    recurse_nodes(t, visited) for t in task.upstream_list
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/www/views.py", line 1483, in <listcomp>
    if node_count[0] < node_limit or t not in visited]
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/www/views.py", line 1478, in recurse_nodes
    visited.add(task)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/models/__init__.py", line 2304, in __hash__
    hash(val)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/models/__init__.py", line 2304, in __hash__
    hash(val)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/models/__init__.py", line 2304, in __hash__
    hash(val)
  [Previous line repeated 477 more times]
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/models/__init__.py", line 2302, in __hash__
    val = getattr(self, c, None)
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/models/__init__.py", line 2396, in dag_id
    if self.has_dag():
  File "/home/admin/.pyenv/versions/3.7.3/lib/python3.7/site-packages/airflow/models/__init__.py", line 2392, in has_dag
    return getattr(self, '_dag', None) is not None
RecursionError: maximum recursion depth exceeded while calling a Python object

config_json.json

{
  "platform_config": {
    "etl_type": "derived_tables_presto|derived_tables_spark",
    "etl_name": "my_company_order_history",
    "version": "3.1",
    "depends_on": [
      "logistics_order_history"
    ]
  },
  "airflow_config": {
    "owner": "my_team",
    "start_date": "2019-12-07",
    "schedule_interval": "* * * 7 *",
    "email": "my_team@my_company.com"
  },
  "job_config": {
    "space": "my_team_derived",
    "table_name": "my_company_order_history",
    "has_dt_params": true,
    "is_dt_partitioned": false,
    "daily_refresh_days": 2,
    "max_refresh_days": 7,
    "etl_sql_params": {
      "k1": "v1"
    }
  }
}

task_order_code.py

@overrides
def build_etl_tasks(self,
                    dag: DAG,
                    etl_config: LocalEtlConfig) -> Tuple[List[BaseOperator], List[BaseOperator]]:
    _, etl_name = etl_config.get_etl_type_name()
    _, _, hive_pool = self.workflow_config.get_hiveserver2_cluster_details()
    _, _, presto_pool = self.workflow_config.get_presto_cluster_details()
    publish_dates_task: PublishDatesOperator = PublishDatesOperator(
        dag=dag,
        task_id=LocalConstants.PUBLISH_DATE_TASK_TMPLT.format(etl_name=etl_name),
        num_days=etl_config.job_config.daily_refresh_days)
    check_or_create_table_schema_task: PythonOperator = PythonOperator(
        dag=dag,
        task_id=LocalConstants.CHECK_TABLE_SCHEMA_TASK_TMPLT.format(etl_name=etl_name),
        python_callable=check_or_create_table_schema_operator.check_or_create_table_schema,
        op_kwargs={LocalConstants.ETL_CONFIG: etl_config, LocalConstants.WORKFLOW_CONFIG: self.workflow_config},
        provide_context=True,
        pool=hive_pool)
    write_interim_data_task: WriteInterimDataOperator = WriteInterimDataOperator(
        dag=dag,
        task_id=LocalConstants.WRITE_INTERIM_DATA_TASK_TMPLT.format(etl_name=etl_name),
        etl_config=etl_config,
        workflow_config=self.workflow_config,
        pool=presto_pool)
    write_final_data_task: PythonOperator = PythonOperator(
        dag=dag,
        task_id=LocalConstants.WRITE_FINAL_DATA_TASK_TMPLT.format(etl_name=etl_name),
        python_callable=write_final_data_operator.write_final_data,
        op_kwargs={LocalConstants.ETL_CONFIG: etl_config, LocalConstants.WORKFLOW_CONFIG: self.workflow_config},
        provide_context=True)
    publish_dates_task >> \
    check_or_create_table_schema_task >> \
    write_interim_data_task >> \
    write_final_data_task
    return ([publish_dates_task], [write_final_data_task])