Skip to content

DagRun.find() fails when given execution_date obtained from context.get('execution_date') directly #32804

Description

@gecko655

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

Airflow version

2.5.3+composer (The latest available airflow version in Cloud Composer)

What happened

DagRun.find() (the find method of SQLAlchemy model DagRun ) fails with the error message TypeError: 'DateTime' object is not iterable, when passing execution_date that is directly obtained from context:

def delete_previous_dagrun_func(dag_to_delete, **context):
    execution_date = context.get('execution_date')
    dagruns_today = DagRun.find(dag_id=dag_to_delete.dag_id,
                                   execution_date=execution_date)

What is the cause

Upon closer inspection, execution_date is of type lazy_object_proxy.Proxy and the is_container() function used in DagRun.find() determines if the variable is a "container" by the presence of the __iter__ field.
lazy_object_proxy.Proxy has __iter__, so the execution_date is determined to be a container, and as a result it is passed as an array element to SQLAlchemy, which caused the "not iterable" error.

if is_container(execution_date):
qry = qry.where(cls.execution_date.in_(execution_date))
elif execution_date is not None:
qry = qry.where(cls.execution_date == execution_date)

def is_container(obj: Any) -> bool:
"""Test if an object is a container (iterable) but not a string."""
return hasattr(obj, "__iter__") and not isinstance(obj, str)

I think is_container() should have another conditional branch to deal with lazy_object_proxy.Proxy.

workaround

It works fine by unwrapping lazy_object_proxy.Proxy before passing it.

def delete_previous_dagrun_func(dag_to_delete, **context):
    execution_date = context.get('execution_date')
    dagruns_today = DagRun.find(dag_id=dag_to_delete.dag_id,
                                   execution_date=execution_date.__wrapped__)

https://github.com/ionelmc/python-lazy-object-proxy/blob/c56c68bda23b8957abbc2fef3d21f32dd44b7f93/src/lazy_object_proxy/simple.py#L76-L83

What you think should happen instead

The variable fetched from context.get('execution_date') can be used directly as an argument of DagRun.find()

How to reproduce

dag_to_delete = DAG('DAG_TO_DELETE')
dag = DAG('DAG')


def delete_previous_dagrun_func(dag_to_delete, **context):
    execution_date = context.get('execution_date')
    dagruns_today = DagRun.find(dag_id=dag_to_delete.dag_id,
                                   execution_date=execution_date)
    # do something for `dagruns_today` (Delete the dagruns for today)


op = PythonOperator(
    task_id='Delete_Previous_DagRun',
    python_callable=delete_previous_dagrun_func,
    op_args=[dag_to_delete],
    provide_context=True,
    dag=dag)

This code produces the following error at DagRun.find line:

[2023-07-24, 19:09:06 JST] {taskinstance.py:1778} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/operators/python.py", line 210, in execute
    branch = super().execute(context)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/operators/python.py", line 175, in execute
    return_value = self.execute_callable()
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/operators/python.py", line 192, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/home/airflow/gcs/dags/*******.py", line ***, in delete_previous_dagrun_func
    dagruns_today = DagRun.find(dag_id=dag_to_delete.dag_id,
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/dagrun.py", line 386, in find
    qry = qry.filter(cls.execution_date.in_(execution_date))
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/operators.py", line 641, in in_
    return self.operate(in_op, other)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/orm/attributes.py", line 317, in operate
    return op(self.comparator, *other, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/operators.py", line 1423, in in_op
    return a.in_(b)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/operators.py", line 641, in in_
    return self.operate(in_op, other)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/orm/properties.py", line 426, in operate
    return op(self.__clause_element__(), *other, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/operators.py", line 1423, in in_op
    return a.in_(b)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/operators.py", line 641, in in_
    return self.operate(in_op, other)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 870, in operate
    return op(self.comparator, *other, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/operators.py", line 1423, in in_op
    return a.in_(b)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/operators.py", line 641, in in_
    return self.operate(in_op, other)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/type_api.py", line 1373, in operate
    return super(TypeDecorator.Comparator, self).operate(
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/type_api.py", line 77, in operate
    return o[0](self.expr, op, *(other + o[1:]), **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/default_comparator.py", line 159, in _in_impl
    seq_or_selectable = coercions.expect(
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/coercions.py", line 193, in expect
    resolved = impl._literal_coercion(
  File "/opt/python3.8/lib/python3.8/site-packages/sqlalchemy/sql/coercions.py", line 573, in _literal_coercion
    element = list(element)
TypeError: 'DateTime' object is not iterable

Operating System

Google Cloud Composer (Ubuntu 20.04.6 LTS on Kubernetes)

Versions of Apache Airflow Providers

No relevant providers

Deployment

Google Cloud Composer

Deployment details

Google Cloud Composer image version: composer-2.3.4-airflow-2.5.3

Anything else

I have recently begun preparing to upgrade Airflow from 1.10 to Series 2.x.
The code described in the reproduce section still works in Airflow 1.10 environment.
I want to know it is intentional OR accidental incompatibility between 1.10 and 2.x.
(If it is intentional, Adding more helpful error message should save time to resolve.)

I am willing to submit a PR, if it is super easy (like changing 1 line of code).
If it is not the case and I need to make a big change, I think I do not have enough time to make PR timely.

Are you willing to submit PR?

  • Yes I am willing to submit a PR!

Code of Conduct

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions