LatestOnlyOperator fails in airflow-providers-standard >=1.1.0

[Limess]

Apache Airflow Provider(s)

standard

Versions of Apache Airflow Providers

apache-airflow-providers-standard==1.1.0

Apache Airflow version

2.10.5, 2.11.0

Operating System

Airflow docker

Deployment

Other Docker-based deployment

Deployment details

Airflow base docker image with minor customisations on EKS

What happened

When upgrading airflow-providers-standard to from 1.0.0 to 1.1.0 or 1.2.0 we receive the following failure + stacktrace at runtime

[2025-05-21, 16:35:42 BST] {taskinstance.py:3313} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 768, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 734, in _execute_callable
    return ExecutionCallableRunner(
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/operator_helpers.py", line 252, in run
    return self.func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 424, in wrapper
    return func(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/standard/operators/branch.py", line 105, in execute
    return self.do_branch(context, self.choose_branch(context))
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/standard/operators/latest_only.py", line 69, in choose_branch
    dates = self._get_compare_dates(dag_run)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/standard/operators/latest_only.py", line 110, in _get_compare_dates
    if prev_info := self.dag.timetable.next_dagrun_info(
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/timetables/interval.py", line 105, in next_dagrun_info
    align_last_data_interval_end = self._align_to_prev(last_automated_data_interval.end)
                                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/timetables/_cron.py", line 145, in _align_to_prev
    prev_time = self._get_prev(current)
                ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/timetables/_cron.py", line 124, in _get_prev
    return convert_to_utc(current.in_timezone(self._timezone) - delta)
                          ^^^^^^^^^^^^^^^^^^^
AttributeError: 'datetime.datetime' object has no attribute 'in_timezone'. Did you mean: 'astimezone'?

The dag definition has the following parameters:

        {
            "start_date": datetime(2019, 9, 1),
            "retries": 2,
            "retry_delay": timedelta(minutes=5),
            "catchup": False,
            "max_active_tasks": 1,
            "default_args": default_args,
            "max_active_runs": 1,
            "schedule": "@hourly",
            "tags": [
                "analytics-ingest",
                "postgres-ingest",
                *dag_options.get("tags", []),
            ],
        },

What you think should happen instead

No error should occur

How to reproduce

Due to the simplicity of the DAG I'm inclined to say this occurs on any airflow 2.10.5/2.11.0 installation using airflow-providers-standard==1.2.0 where schedule="@hourly" and LatestOnlyOperator is used,

e.g.

from airflow.providers.standard.operators.latest_only import LatestOnlyOperator

LatestOnlyOperator(task_id="only_run_latest_execution")

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[leakec]

I'm experiencing the same issue in 3.0.1.

In airflow/providers/standard/operators/latest_only.py near line 102 I see

        current_interval = DataInterval(
            start=dag_run.data_interval_start or dagrun_date,
            end=dag_run.data_interval_end or dagrun_date,
        )

I think the dag_run.data_interval_start and dag_run.data_interval_end may be datetime objects (from the datetime package) rather than pendulum.DateTime objects, which could be leading to this error.

[leakec]

Converting these lines to a pendulum.instance as shown below fixed the issue for me:

        current_interval = DataInterval(
            start=pendulum.instance(dag_run.data_interval_start) or dagrun_date,
            end=pendulum.instance(dag_run.data_interval_end) or dagrun_date,
        )

[leakec]

I tried to re-create this by adding a test in airflow/providers/standard/tests/unit/standard/operators/test_latest_only_operator.py like

    def test_regular_latest_only_run(self, run_task):
        """Test latest_only running in normal mode."""
        with DAG(
            "test_dag",
            start_date=datetime.datetime(2024, 12, 11),
            schedule="0 * * * *",
            catchup=False,
        ):
            latest_task = LatestOnlyOperator(task_id="latest")
            downstream_task = EmptyOperator(task_id="downstream")
            latest_task >> downstream_task

        run_task(latest_task, run_type=DagRunType.SCHEDULED)

        assert run_task.dagrun.data_interval_start != run_task.dagrun.data_interval_end

        # The task will raise DownstreamTasksSkipped exception if it is not the latest run
        assert run_task.state == State.SUCCESS

but I could not trigger the same error. For some reason in this case the value is passed as an actual DateTime rather than a datetime. I think this is likely something I'm doing incorrectly in setting up the test case, but I don't know enough about how airflow testing works to figure out what, if anything, is different.

I do know that this needs to occur when a schedule is a cron trigger running every hour to make it to the portion of _cron.py that tries to call in_timezone on the incoming value. However, for some reason in the test case it has the proper type of DateTime.

I'm hoping someone with more knowledge of airflow testing can pick up where I left off, or help me fix any mistakes I have made.

EDIT: I've tried the cron trigger as "0 * * * * " and " * * * * *", but neither seems to error the test out.

[leakec]

Stepped up to airflow 3.0.3 and airflow-providers-standard 1.4.0, but this is still an issue for me.

[potiuk]

Stepped up to airflow 3.0.3 and airflow-providers-standard 1.4.0, but this is still an issue for me.

Which means? What is your problem in detail? can you describe it and provide YOUR stack traces with those versions? That halps anyne who tries to analyse the issue in their free time to do it as efficiently as possible, and it's actually about a million times more helpful than adding a comment "I have the same problem"

[leakec]

@potiuk My apologies, you're right, I should have included them. Here is what I'm seeing in airflow. I've stripped out absolute paths so they are relative to the airflow installed directory.

Task failed with exception: source="task"
AttributeError: 'datetime.datetime' object has no attribute 'in_timezone'
File "lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 877 in run
File "lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1164 in _execute_task
File "lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 397 in wrapper
File "lib/python3.12/site-packages/airflow/providers/standard/operators/branch.py", line 104 in execute
File "lib/python3.12/site-packages/airflow/providers/standard/operators/latest_only.py", line 69 in choose_branch
File "lib/python3.12/site-packages/airflow/providers/standard/operators/latest_only.py", line 110 in _get_compare_dates
File "lib/python3.12/site-packages/airflow/timetables/interval.py", line 105 in next_dagrun_info
File "lib/python3.12/site-packages/airflow/timetables/_cron.py", line 149 in _align_to_prev
File "lib/python3.12/site-packages/airflow/timetables/_cron.py", line 128 in _get_prev

EDIT: Edited stack trace above to reflect the test dag shown down below.

It's also worth noting I fixed this locally using the same fix I mentioned above. I would be happy to make a PR for this, but I got stuck on re-creating the issue in testing (see my comment above). If you have any suggestions there, I'd be happy to try them. Thanks!

[potiuk]

I guess then that might be somethig @uranusjr might be best to answer. If that clearly reproducible in latest version of airflow - I marked it as "affected_version:3.0"

[leakec]

For anyone looking to reproduce the problem, a simple way to do so is run the following DAG in an airflow instance:

from airflow import DAG
from datetime import datetime
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.latest_only import LatestOnlyOperator

with DAG(
    "test_job",
    description="Job to test latest only operator.",
    schedule="* * * * *",
    start_date=datetime(2024, 12, 11),
    catchup=False,
):

    # latest only operator
    task1 = LatestOnlyOperator(task_id="latest_only")
    task2 = BashOperator(bash_command="echo test", task_id="task2")

    task1 >> task2

[gopidesupavan]

am able to reproduce this, i think the error seems to happen only when there is cron expression provided, likely _get_next getting datatime.datetime type , instead of pendulum.DateTime. IMO its better to update this function to convert pendulum the current function arg value.

airflow/airflow-core/src/airflow/timetables/_cron.py

Line 106 in 4eb776a

def _get_next(self, current: DateTime) -> DateTime:

[gopidesupavan]

@leakec would like to open PR ?