Move TriggerDagRunOperator to standard provider (apache#44053)

* Moved trigger_dagrun operator to standard provider

* Changed all imports for dagrun_operator from operators to standard provider.

* Changed all imports for dagrun_operator from operators to standard provider.

* Update trigger_dagrun doc reference

* Revert Changes

* Add trigger_dagrun to standard provider.yaml file.

* Change BaseOperator[Link] import path.

airflow/example_dags/example_trigger_controller_dag.py

@@ -26,7 +26,7 @@
 import pendulum
 
 from airflow.models.dag import DAG
-from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
 
 with DAG(
     dag_id="example_trigger_controller_dag",

airflow/serialization/serialized_objects.py

@@ -119,7 +119,7 @@
 log = logging.getLogger(__name__)
 
 _OPERATOR_EXTRA_LINKS: set[str] = {
-    "airflow.operators.trigger_dagrun.TriggerDagRunLink",
+    "airflow.providers.standard.operators.trigger_dagrun.TriggerDagRunLink",
     "airflow.sensors.external_task.ExternalDagLink",
     # Deprecated names, so that existing serialized dags load straight away.
     "airflow.sensors.external_task.ExternalTaskSensorLink",
@@ -1020,7 +1020,7 @@ class DependencyDetector:
     @staticmethod
     def detect_task_dependencies(task: Operator) -> list[DagDependency]:
         """Detect dependencies caused by tasks."""
-        from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+        from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
         from airflow.sensors.external_task import ExternalTaskSensor
 
         deps = []

docs/apache-airflow/core-concepts/dags.rst

@@ -765,7 +765,7 @@ While dependencies between tasks in a DAG are explicitly defined through upstrea
 relationships, dependencies between DAGs are a bit more complex. In general, there are two ways
 in which one DAG can depend on another:
 
-- triggering - :class:`~airflow.operators.trigger_dagrun.TriggerDagRunOperator`
+- triggering - :class:`~airflow.providers.standard.operators.trigger_dagrun.TriggerDagRunOperator`
 - waiting - :class:`~airflow.sensors.external_task_sensor.ExternalTaskSensor`
 
 Additional difficulty is that one DAG could wait for or trigger several runs of the other DAG

docs/apache-airflow/operators-and-hooks-ref.rst

@@ -65,7 +65,7 @@ For details see: :doc:`apache-airflow-providers:operators-and-hooks-ref/index`.
    * - :mod:`airflow.operators.latest_only`
      -
 
-   * - :mod:`airflow.operators.trigger_dagrun`
+   * - :mod:`airflow.providers.standard.operators.trigger_dagrun`
      -
 
 **Sensors:**

airflow/operators/trigger_dagrun.py → providers/src/airflow/providers/standard/operators/trigger_dagrun.py

@@ -34,8 +34,7 @@
     DagNotFound,
     DagRunAlreadyExists,
 )
-from airflow.models.baseoperator import BaseOperator
-from airflow.models.baseoperatorlink import BaseOperatorLink
+from airflow.models import BaseOperator, BaseOperatorLink
 from airflow.models.dag import DagModel
 from airflow.models.dagbag import DagBag
 from airflow.models.dagrun import DagRun

providers/src/airflow/providers/standard/provider.yaml

@@ -48,6 +48,7 @@ operators:
       - airflow.providers.standard.operators.bash
       - airflow.providers.standard.operators.python
       - airflow.providers.standard.operators.generic_transfer
+      - airflow.providers.standard.operators.trigger_dagrun
 
 sensors:
   - integration-name: Standard

tests/operators/test_trigger_dagrun.py

@@ -30,7 +30,7 @@
 from airflow.models.dagrun import DagRun
 from airflow.models.log import Log
 from airflow.models.taskinstance import TaskInstance
-from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
 from airflow.settings import TracebackSessionForTests
 from airflow.triggers.external_task import DagStateTrigger
 from airflow.utils import timezone
@@ -107,7 +107,9 @@ def assert_extra_link(self, triggered_dag_run, triggering_task, session):
             )
             .one()
         )
-        with mock.patch("airflow.operators.trigger_dagrun.build_airflow_url_with_query") as mock_build_url:
+        with mock.patch(
+            "airflow.providers.standard.operators.trigger_dagrun.build_airflow_url_with_query"
+        ) as mock_build_url:
             triggering_task.get_extra_links(triggering_ti, "Triggered DAG")
         assert mock_build_url.called
         args, _ = mock_build_url.call_args

tests/serialization/test_dag_serialization.py

@@ -1806,7 +1806,7 @@ def test_derived_dag_deps_operator(self):
         Tests DAG dependency detection for operators, including derived classes
         """
         from airflow.operators.empty import EmptyOperator
-        from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+        from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
 
         class DerivedOperator(TriggerDagRunOperator):
             pass

tests/system/core/example_external_task_parent_deferrable.py

@@ -18,7 +18,7 @@
 
 from airflow import DAG
 from airflow.operators.empty import EmptyOperator
-from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
 from airflow.sensors.external_task import ExternalTaskSensor
 from airflow.utils.timezone import datetime