Some branch operators not working when in manually triggered DAG runs due to `KeyError: 'logical_date'`

[ramitkataria]

Apache Airflow version

main (development)

If "Other Airflow 2 version" selected, which one?

No response

What happened?

These example DAGs for branch operators fail when the DAG is manually triggered:

• The dag example_branch_datetime_operator_3 in example_branch_datetime_operator.py fails when the task datetime_branch fails with this stack trace:

[2025-03-27, 20:59:40] INFO - DAG bundles loaded: dags-folder: source="***.dag_processing.bundles.manager.DagBundlesManager"

[2025-0[3](http://localhost:28080/dags/example_branch_datetime_operator_3/runs/manual__2025-03-27T20:59:40.510906+00:00_LdHU1Uhe/tasks/datetime_branch?try_number=1#3)-27, 20:59:40] INFO - Filling up the DagBag from /files/dags/example_branch_datetime_operator.py: source="***.models.dagbag.DagBag"

[2025-03-27, 20:59:[4](http://localhost:28080/dags/example_branch_datetime_operator_3/runs/manual__2025-03-27T20:59:40.510906+00:00_LdHU1Uhe/tasks/datetime_branch?try_number=1#4)0] ERROR - Task failed with exception: source="task"
KeyError: 'logical_date'
File "/opt/***/task-sdk/src/***/sdk/execution_time/task_runner.py", line 646 in run

File "/opt/***/task-sdk/src/***/sdk/execution_time/task_runner.py", line 903 in _execute_task

File "/opt/***/task-sdk/src/***/sdk/definitions/baseoperator.py", line 380 in wrapper

File "/opt/***/providers/standard/src/***/providers/standard/operators/branch.py", line 105 in execute

File "/opt/***/providers/standard/src/***/providers/standard/operators/datetime.py", line 80 in choose_branch

• The dag example_branch_dop_operator_v3 fails when the task should_run fails with this stack trace:

[2025-03-27, 21:02:23] INFO - DAG bundles loaded: dags-folder: source="***.dag_processing.bundles.manager.DagBundlesManager"

[2025-0[3](http://localhost:28080/dags/example_branch_dop_operator_v3/runs/manual__2025-03-27T21:02:22.741244+00:00_QDv3CMdV/tasks/should_run?try_number=1#3)-27, 21:02:23] INFO - Filling up the DagBag from /files/dags/example_branch_python_dop_operator_3.py: source="***.models.dagbag.DagBag"

[2025-03-27, 21:02:23] ERROR - Task failed with exception: source="task"
KeyError: 'logical_date'
File "/opt/***/task-sdk/src/***/sdk/execution_time/task_runner.py", line 6[4](http://localhost:28080/dags/example_branch_dop_operator_v3/runs/manual__2025-03-27T21:02:22.741244+00:00_QDv3CMdV/tasks/should_run?try_number=1#4)6 in run

File "/opt/***/task-sdk/src/***/sdk/execution_time/task_runner.py", line 903 in _execute_task

File "/opt/***/task-sdk/src/***/sdk/definitions/baseoperator.py", line 380 in wrapper

File "/opt/***/***-core/src/***/decorators/base.py", line 252 in execute

File "/opt/***/task-sdk/src/***/sdk/definitions/baseoperator.py", line 380 in wrapper

File "/opt/***/providers/standard/src/***/providers/standard/operators/python.py", line 255 in execute

File "/opt/***/task-sdk/src/***/sdk/definitions/baseoperator.py", line 380 in wrapper

File "/opt/***/providers/standard/src/***/providers/standard/operators/python.py", line 212 in execute

File "/opt/***/providers/standard/src/***/providers/standard/operators/python.py", line 235 in execute_callable

File "/opt/***/task-sdk/src/***/sdk/execution_time/callback_runner.py", line 81 in run

File "/files/dags/example_branch_python_dop_operator_3.py", line 40 in should_run

What you think should happen instead?

These tasks are not supposed to fail just like what happens in scheduled runs. They are failing because of KeyError: 'logical_date'

How to reproduce

• Copy example DAGs in files example_branch_python_dop_operator_3.py and example_branch_datetime_operator to your dags directory
• Enable DAGs example_branch_datetime_operator_3 and example_branch_dop_operator_v3
• Trigger a DAG run in each of them

Operating System

macOS 15.3.2

Versions of Apache Airflow Providers

Main branch, commit 64aa6fea179926016630ca94ac8421d2e5692997

Deployment

Other

Deployment details

Using breeze with command: breeze start-airflow --backend postgres -f --python 3.9

Anything else?

Problem occurs every time

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[ramitkataria]

cc @shahar1 @kaxil

[gopidesupavan]

ah this is looking for logical_date in context, we dont have that we need to use some thing different there, like run_after

airflow/providers/standard/src/airflow/providers/standard/operators/datetime.py

Line 80 in 347c98c

now = context["logical_date"]

[ferruzzi]

Thanks for looking into this and opening the Issue, Ramit.

[ramitkataria]

@gopidesupavan It looks like the first example (example_branch_datetime_operator_3) is fixed but the second one (example_branch_dop_operator_v3) still fails with a similar stack trace:

[2025-04-01, 00:23:55] INFO - DAG bundles loaded: dags-folder: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-04-01, 00:2[3](http://localhost:28080/dags/example_branch_dop_operator_v3/runs/manual__2025-04-01T00:23:54.932689+00:00_zWc8U2FN/tasks/should_run?try_number=1#3):55] INFO - Filling up the DagBag from /files/dags/example_branch_python_dop_operator_3.py: source="airflow.models.dagbag.DagBag"
[2025-04-01, 00:23:55] ERROR - Task failed with exception: source="task"
KeyError: 'logical_date'
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 670 in run

File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 927 in _execute_task

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 380 in wrapper

File "/opt/airflow/airflow-core/src/airflow/decorators/base.py", line 252 in execute

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 380 in wrapper

File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py", line 255 in execute

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 380 in wrapper

File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py", line 212 in execute

File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py", line 235 in execute_callable

File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line 81 in run

File "/files/dags/example_branch_python_dop_operator_3.py", line [4](http://localhost:28080/dags/example_branch_dop_operator_v3/runs/manual__2025-04-01T00:23:54.932689+00:00_zWc8U2FN/tasks/should_run?try_number=1#4)0 in should_run

Does it work on your machine?

[gopidesupavan]

Yeah this example slightly wont work, the example explicitly using logical_date and doing some calculation https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/example_dags/example_branch_python_dop_operator_3.py#L41, now that logical_date field is not populating in context as its empty, so it wont work as expected with AF 2. basically the example needs updating.