Fix: Issue when the database objects passed to the cleanup task via params aren't JSON serializable

[ealebed]

Issue link

[lonyelon]

Great pull request! 100% solves the problem!

[Limess]

It'd be great to see this in merged if possible - this is holding us up on upgrading airflow (we could point to this PR but would prefer not to)

[hanizaidi110]

Hey guys,

I'm getting this error:

Task failed with an exception Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1332, in _run_raw_task self._execute_task_with_callbacks(context) File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1458, in _execute_task_with_callbacks result = self._execute_task(context, self.task) File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1514, in _execute_task result = execute_callable(context=context) File "/usr/local/lib/python3.9/site-packages/airflow/operators/python.py", line 151, in execute return_value = self.execute_callable() File "/usr/local/lib/python3.9/site-packages/airflow/operators/python.py", line 162, in execute_callable return self.python_callable(*self.op_args, **self.op_kwargs) File "/vimcar-bi/airflow_dags/admin_airflow_db_cleanup.py", line 275, in cleanup_function airflow_db_model = DATABASE_OBJECTS_DICTS[object_name].get("airflow_db_model") KeyError: 'TaskSet' Marking task as FAILED. dag_id=admin_airflow_db_cleanup, task_id=cleanup_TaskSet, execution_date=20211201T172422, start_date=20211201T172533, end_date=20211201T172533

[pceric]

This doesn't work if you aren't using the CeleryExecutor as Task and TaskSet don't exist. There needs to be a conditional check like on line 247.