Repository: apache/airflow
Pull request: Make default as time.now() for TriggerDagRunOperator
closes: https://github.com/apache/airflow/issues/48412
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).


Problem statement:
Manually triggered dags should have default logical_date and run_after of "right now"
### Body

As part of the AIP-83 amendment process we decided to make logical_date nullable, and to make it default null for manually triggered dags.

We got some feedback that this will suck for users.

So we want to keep the airlfow 2 behavior where logical_date is `now()` by default, but the users can optionally select null.

**NOTE: This will have a UI component!!!** 

We will need to update the UI date picker so that user can optionally choose null.


1. Manually triggering DAG Run via UI
* date picker will default to the time when modal opened
* user can null it out somehow
* bonus if there's a button to clear the date
2. Triggering Dag Run via API
* try to make it so if not supplied then it will be timezone.now() but if supplied we respect it. if this causes trouble and is more or less impossible to support (1), then we can keep it required
4. Trigger Dag Run Operator usage* if unset, it should take the logical date (yes, even if nullable) from the triggering dag run
* if set, respect it no matter what



### Committer

- [x] I acknowledge that I am a maintainer/committer of the Apache Airflow project.

Hints from issue discussion:
Three scenarios we need to account for are:
1. Manually triggering DAG Run via UI
2. Triggering Dag Run via API
3. Trigger Dag Run Operator usage
@vatsrahul1001 talk to me / constance to confirm requirements i was going to review the slack thread again
@dstandish / @cmarteepants Currently, logical_date is a required field in the API request; however, the user can pass it as null.

Do we want to make logical_date optional, defaulting to datetime.now() if not provided, while respecting any value explicitly passed?
> Triggering Dag Run via API
> try to make it so if not supplied then it will be timezone.now() but if supplied we respect it. if this causes trouble and is more or less impossible to support (1), then we can keep it required 
Currently, for `TriggerDagRunOperator`, logical_date defaults to None if not provided. Do we want to change this to datetime.now() as the default?
Just started a lazy consensus vote: https://lists.apache.org/list.html?dev@airflow.apache.org
cc @uranusjr 
I think it makes the most sense for TriggerDagRunOperator to default to `now()` if nothing is explicitly supplied. This is the current behaviour, and anyone using the operator is likely already OK with it. If the possibility of supplying *None* benefits some, they can opt into it explicitly.

“Not supplying a value” does not *technically* apply to the UI since the date field is always there and always included in the request by the frontend. A more technically correct way to put it is what initial value we present to the user in the form. As long as there is an easy way to set the value to “now” or “empty” (instead of needing to manually click through the datetimepicker, which is a lot of work), either default should be fine. Personally I lean a little toward None, but defaulting to now is not a big deal either since the possibility of a conflict should be next to none (pun unintended).

Personally I feel the API should not have a default at all, but this opinion was not received well when I proposed in Slack. If we must have a default, it should be now to match TriggerDagRunOperator. I want to state again that I still think not having a default is the correct way to go since the number of users relying on the API should be relatively low, and supplying a “now” value in the API is comparatively easy. Defaulting to now is very unintuitive and confusing to new users IMO, but it is the correct way to go if the goal is to break as few existing users as possible at any cost to potential new users,.
I feel strongly about TriggerDagRunOperator and the UI, and glad we're on the same page. If that's all we update, I can live with that. 

For the API, I'm relying on feedback from DevRel that we are underestimating API usage and concern about the quantity of things that need to be changed (even if most of the changes are easy). I agree with you that if you are starting fresh, it's awkward to replace "now" with None though, if you fall in that category. TLDR: I'm personally on the fence re API, but erring towards trusting feedback from DevRel for this. 
Is it too early in the major version lifetime to have a config for this? 😛 
UI: https://github.com/apache/airflow/pull/48690

Commit message: fix sdk tests

Patch to review:
diff --git a/providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py b/providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py
index 24b9a59b49a45..d9b44ac306340 100644
--- a/providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py
+++ b/providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py
@@ -190,10 +190,12 @@ def __init__(
         self.logical_date = logical_date
 
     def execute(self, context: Context):
-        if self.logical_date is None or isinstance(self.logical_date, datetime.datetime):
+        if isinstance(self.logical_date, datetime.datetime):
             parsed_logical_date = self.logical_date
-        else:
+        elif isinstance(self.logical_date, str):
             parsed_logical_date = timezone.parse(self.logical_date)
+        else:
+            parsed_logical_date = timezone.utcnow()
 
         try:
             json.dumps(self.conf)
diff --git a/providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py b/providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py
index 55dc7eccc4dc0..b82b9e0929ee2 100644
--- a/providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py
+++ b/providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py
@@ -108,7 +108,7 @@ def test_trigger_dagrun(self):
 
             assert exc_info.value.trigger_dag_id == TRIGGERED_DAG_ID
             assert exc_info.value.conf == {"foo": "bar"}
-            assert exc_info.value.logical_date is None
+            assert exc_info.value.logical_date is not None
             assert exc_info.value.reset_dag_run is False
             assert exc_info.value.skip_when_already_exists is False
             assert exc_info.value.wait_for_completion is False
@@ -119,7 +119,7 @@ def test_trigger_dagrun(self):
                 run_type=DagRunType.MANUAL, run_after=timezone.utcnow()
             ).rsplit("_", 1)[0]
             # rsplit because last few characters are random.
-            assert exc_info.value.dag_run_id.rsplit("_", 1)[0] == expected_run_id
+            assert exc_info.value.dag_run_id == expected_run_id
 
     @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Implementation is different for Airflow 2 & 3")
     @mock.patch("airflow.providers.standard.operators.trigger_dagrun.XCom.get_one")
diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
index 6b25bb81bf20f..1db6efa1b14cf 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
@@ -117,6 +117,7 @@
 
 if TYPE_CHECKING:
     from kgb import SpyAgency
+import time_machine
 
 
 def get_inline_dag(dag_id: str, task: BaseOperator) -> DAG:
@@ -2225,6 +2226,7 @@ class CustomOperator(BaseOperator):
 class TestTriggerDagRunOperator:
     """Tests to verify various aspects of TriggerDagRunOperator"""
 
+    @time_machine.travel("2025-01-01 00:00:00", tick=False)
     def test_handle_trigger_dag_run(self, create_runtime_ti, mock_supervisor_comms):
         """Test that TriggerDagRunOperator (with default args) sends the correct message to the Supervisor"""
         from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
@@ -2249,6 +2251,7 @@ def test_handle_trigger_dag_run(self, create_runtime_ti, mock_supervisor_comms):
                     dag_id="test_dag",
                     run_id="test_run_id",
                     reset_dag_run=False,
+                    logical_date=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
                 ),
                 log=mock.ANY,
             ),
@@ -2274,6 +2277,7 @@ def test_handle_trigger_dag_run(self, create_runtime_ti, mock_supervisor_comms):
             (False, TaskInstanceState.FAILED),
         ],
     )
+    @time_machine.travel("2025-01-01 00:00:00", tick=False)
     def test_handle_trigger_dag_run_conflict(
         self, skip_when_already_exists, expected_state, create_runtime_ti, mock_supervisor_comms
     ):
@@ -2299,6 +2303,7 @@ def test_handle_trigger_dag_run_conflict(
             mock.call.send_request(
                 msg=TriggerDagRun(
                     dag_id="test_dag",
+                    logical_date=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
                     run_id="test_run_id",
                     reset_dag_run=False,
                 ),
@@ -2318,6 +2323,7 @@ def test_handle_trigger_dag_run_conflict(
             ([DagRunState.SUCCESS], None, DagRunState.FAILED, DagRunState.FAILED),
         ],
     )
+    @time_machine.travel("2025-01-01 00:00:00", tick=False)
     def test_handle_trigger_dag_run_wait_for_completion(
         self,
         allowed_states,
@@ -2367,6 +2373,7 @@ def test_handle_trigger_dag_run_wait_for_completion(
                 msg=TriggerDagRun(
                     dag_id="test_dag",
                     run_id="test_run_id",
+                    logical_date=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
                 ),
                 log=mock.ANY,
             ),


Code review comments:
diff_hunk: @@ -190,10 +190,12 @@ def __init__(
         self.logical_date = logical_date
 
     def execute(self, context: Context):
-        if self.logical_date is None or isinstance(self.logical_date, datetime.datetime):
+        if isinstance(self.logical_date, datetime.datetime):
             parsed_logical_date = self.logical_date
-        else:
+        elif isinstance(self.logical_date, str):
             parsed_logical_date = timezone.parse(self.logical_date)
+        else:
+            parsed_logical_date = timezone.utcnow()
path: providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py
text: @user1:
This removes the possibility to explicitly supply None as the logical date. We need an extra marker, something like

```
def __init__(self, ..., logical_date: str | datetime | None | ArgNotSet = NOT_SET):
    ...

def execute(self, context):
    if self.logical_date is None or isinstance(self.logical_date, datetime):
        parsed_logical_date = self.logical_date
    elif isinstance(self.logical_date, str):
        parsed_logical_date = timezone.parse(self.logical_date)
    else:
        parsed_logical_date = utcnow()
```