apache · amoghrajesh · Mar 7, 2025 · Feb 24, 2025 · Feb 24, 2025 · Feb 24, 2025
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -694,10 +694,11 @@ repos:
       - id: check-base-operator-usage
         language: pygrep
         name: Check BaseOperatorLink core imports
-        description: Make sure BaseOperatorLink is imported from airflow.models.baseoperatorlink in core
-        entry: "from airflow\\.models import.* BaseOperatorLink"
+        description: Make sure BaseOperatorLink is not imported from airflow.models in core
+        entry: "^\\s*from airflow\\.models\\.baseoperatorlink import BaseOperatorLink\\b"
         files: \.py$
         pass_filenames: true
+        echo: True
         exclude: >
           (?x)
           ^airflow/decorators/.*$|
@@ -708,9 +709,9 @@ repos:
           ^dev/provider_packages/.*$
       - id: check-base-operator-usage
         language: pygrep
-        name: Check BaseOperator[Link] other imports
-        description: Make sure BaseOperator[Link] is imported from airflow.models outside of core
-        entry: "from airflow\\.models\\.baseoperator(link)? import.* BaseOperator"
+        name: Check BaseOperator other imports
+        description: Make sure BaseOperator is imported from airflow.models outside of core
+        entry: "from airflow\\.models\\.baseoperator import.* BaseOperator"
         pass_filenames: true
         files: >
           (?x)

diff --git a/airflow/models/__init__.py b/airflow/models/__init__.py
@@ -88,7 +88,7 @@ def __getattr__(name):
     "ID_LEN": "airflow.models.base",
     "Base": "airflow.models.base",
     "BaseOperator": "airflow.models.baseoperator",
-    "BaseOperatorLink": "airflow.models.baseoperatorlink",
+    "BaseOperatorLink": "airflow.sdk.definitions.baseoperatorlink",
     "Connection": "airflow.models.connection",
     "DagBag": "airflow.models.dagbag",
     "DagModel": "airflow.models.dag",
@@ -117,7 +117,6 @@ def __getattr__(name):
     from airflow.jobs.job import Job
     from airflow.models.base import ID_LEN, Base
     from airflow.models.baseoperator import BaseOperator
-    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.connection import Connection
     from airflow.models.dag import DAG, DagModel, DagTag
     from airflow.models.dagbag import DagBag
@@ -137,4 +136,5 @@ def __getattr__(name):
     from airflow.models.trigger import Trigger
     from airflow.models.variable import Variable
     from airflow.models.xcom import XCom
+    from airflow.sdk import BaseOperatorLink
     from airflow.sdk.definitions.param import Param
diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
@@ -88,9 +88,9 @@
     from sqlalchemy.orm import Session
 
     from airflow.models.abstractoperator import TaskStateChangeCallback
-    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.dag import DAG as SchedulerDAG
     from airflow.models.operator import Operator
+    from airflow.sdk import BaseOperatorLink
     from airflow.sdk.definitions.node import DAGNode
     from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
     from airflow.triggers.base import StartTriggerArgs

diff --git a/airflow/serialization/serialized_objects.py b/airflow/serialization/serialized_objects.py
@@ -40,7 +40,6 @@
 from airflow.callbacks.callback_requests import DagCallbackRequest, TaskCallbackRequest
 from airflow.exceptions import AirflowException, SerializationError, TaskDeferred
 from airflow.models.baseoperator import BaseOperator
-from airflow.models.baseoperatorlink import BaseOperatorLink, XComOperatorLink
 from airflow.models.connection import Connection
 from airflow.models.dag import DAG, _get_model_data_interval
 from airflow.models.expandinput import (
@@ -64,6 +63,7 @@
     BaseAsset,
 )
 from airflow.sdk.definitions.baseoperator import BaseOperator as TaskSDKBaseOperator
+from airflow.sdk.definitions.baseoperatorlink import XComOperatorLink
 from airflow.sdk.definitions.mappedoperator import MappedOperator
 from airflow.sdk.definitions.param import Param, ParamsDict
 from airflow.sdk.definitions.taskgroup import MappedTaskGroup, TaskGroup
@@ -98,6 +98,7 @@
 
     from airflow.models import DagRun
     from airflow.models.expandinput import ExpandInput
+    from airflow.sdk import BaseOperatorLink
     from airflow.sdk.definitions._internal.node import DAGNode
     from airflow.sdk.types import Operator
     from airflow.serialization.json_schema import Validator

diff --git a/contributing-docs/08_static_code_checks.rst b/contributing-docs/08_static_code_checks.rst
@@ -132,7 +132,7 @@ require Breeze Docker image to be built locally.
 +-----------------------------------------------------------+--------------------------------------------------------+---------+
 | check-base-operator-usage                                 | * Check BaseOperator core imports                      |         |
 |                                                           | * Check BaseOperatorLink core imports                  |         |
-|                                                           | * Check BaseOperator[Link] other imports               |         |
+|                                                           | * Check BaseOperator other imports                     |         |
 +-----------------------------------------------------------+--------------------------------------------------------+---------+
 | check-boring-cyborg-configuration                         | Checks for Boring Cyborg configuration consistency     |         |
 +-----------------------------------------------------------+--------------------------------------------------------+---------+

diff --git a/devel-common/src/tests_common/test_utils/compat.py b/devel-common/src/tests_common/test_utils/compat.py
@@ -36,7 +36,7 @@
     from airflow.models.errors import ImportError as ParseImportError  # type: ignore[no-redef,attr-defined]
 
 try:
-    from airflow.models.baseoperatorlink import BaseOperatorLink
+    from airflow.sdk import BaseOperatorLink
 except ImportError:
     # Compatibility for Airflow 2.7.*
     from airflow.models.baseoperator import BaseOperatorLink

diff --git a/docs/apache-airflow/howto/define-extra-link.rst b/docs/apache-airflow/howto/define-extra-link.rst
@@ -31,7 +31,7 @@ The following code shows how to add extra links to an operator via Plugins:
 .. code-block:: python
 
     from airflow.models.baseoperator import BaseOperator
-    from airflow.models.baseoperatorlink import BaseOperatorLink
+    from airflow.sdk import BaseOperatorLink
     from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.plugins_manager import AirflowPlugin
 
@@ -85,7 +85,7 @@ tasks using :class:`~airflow.providers.amazon.aws.transfers.gcs_to_s3.GCSToS3Ope
 .. code-block:: python
 
   from airflow.models.baseoperator import BaseOperator
-  from airflow.models.baseoperatorlink import BaseOperatorLink
+  from airflow.sdk import BaseOperatorLink
   from airflow.models.taskinstancekey import TaskInstanceKey
   from airflow.plugins_manager import AirflowPlugin
   from airflow.providers.amazon.aws.transfers.gcs_to_s3 import GCSToS3Operator
@@ -128,7 +128,7 @@ Console, but if we wanted to change that link we could:
 .. code-block:: python
 
     from airflow.models.baseoperator import BaseOperator
-    from airflow.models.baseoperatorlink import BaseOperatorLink
+    from airflow.sdk import BaseOperatorLink
     from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.models.xcom import XCom
     from airflow.plugins_manager import AirflowPlugin

diff --git a/docs/apache-airflow/public-airflow-interface.rst b/docs/apache-airflow/public-airflow-interface.rst
@@ -264,19 +264,15 @@ can be implemented to respond to DAG/Task lifecycle events.
 
 You can read more about Listeners in :doc:`administration-and-deployment/listeners`.
 
+..
+  TODO AIP-72: This class has been moved to task sdk but we cannot add a doc reference for it yet because task sdk doesn't have rendered docs yet.
+
 Extra Links
 -----------
 
 Extra links are dynamic links that could be added to Airflow independently from custom Operators. Normally
 they can be defined by the Operators, but plugins allow you to override the links on a global level.
 
-.. toctree::
-  :includehidden:
-  :glob:
-  :maxdepth: 1
-
-  _api/airflow/models/baseoperatorlink/index
-
 You can read more about the Extra Links in :doc:`/howto/define-extra-link`.
 
 Using Public Interface to integrate with external services and applications

@@ -277,7 +277,6 @@ def _get_rst_filepath_from_path(filepath: pathlib.Path):
 
     models_included: set[str] = {
         "baseoperator.py",
-        "baseoperatorlink.py",
         "connection.py",
         "dag.py",
         "dagrun.py",

diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
@@ -171,6 +171,8 @@ BaseHook
 BaseObject
 BaseOperator
 baseOperator
+BaseOperatorLink
+baseoperatorlink
 basestring
 basetaskrunner
 BaseView

diff --git a/generated/provider_dependencies.json b/generated/provider_dependencies.json
@@ -494,6 +494,7 @@
     "devel-deps": [],
     "plugins": [],
     "cross-providers-deps": [
+      "common.compat",
       "http",
       "openlineage"
     ],
@@ -1404,7 +1405,9 @@
     ],
     "devel-deps": [],
     "plugins": [],
-    "cross-providers-deps": [],
+    "cross-providers-deps": [
+      "common.compat"
+    ],
     "excluded-python-versions": [],
     "state": "ready"
   },

diff --git a/newsfragments/aip-72.significant.rst b/newsfragments/aip-72.significant.rst
@@ -42,6 +42,9 @@ As part of this change the following breaking changes have occurred:
 
   It is recommended that you replace such a custom operator with a deferrable sensor, a condition or another triggering mechanism.
 
+- ``BaseOperatorLink`` has now been moved into the task SDK to be consumed by DAG authors to write custom operator links.
+
+  Any occurrences of imports from ``airflow.models.baseoperatorlink`` will need to be updated to ``airflow.sdk.definitions.baseoperatorlink``
 
 * Types of change
 
@@ -50,7 +53,7 @@ As part of this change the following breaking changes have occurred:
   * [ ] API changes
   * [ ] CLI changes
   * [x] Behaviour changes
-  * [ ] Plugin changes
+  * [x] Plugin changes
   * [ ] Dependency changes
   * [ ] Code interface changes
 
@@ -60,3 +63,9 @@ As part of this change the following breaking changes have occurred:
 
     * [x] ``core.task_runner``
     * [x] ``core.enable_xcom_pickling``
+
+  * ruff
+
+    * AIR302
+
+      * [ ] ``airflow.models.baseoperatorlink`` → ``airflow.sdk``
@@ -19,14 +19,20 @@
 
 from typing import TYPE_CHECKING, ClassVar
 
-from airflow.models import BaseOperatorLink, XCom
+from airflow.models import XCom
 from airflow.providers.amazon.aws.utils.suppress import return_on_error
+from airflow.providers.amazon.version_compat import AIRFLOW_V_3_0_PLUS
 
 if TYPE_CHECKING:
     from airflow.models import BaseOperator
     from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.utils.context import Context
 
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import BaseOperatorLink
+else:
+    from airflow.models.baseoperatorlink import BaseOperatorLink  # type: ignore[no-redef]
+
 
 BASE_AWS_CONSOLE_LINK = "https://console.{aws_domain}"
 

diff --git a/providers/databricks/src/airflow/providers/databricks/operators/databricks.py b/providers/databricks/src/airflow/providers/databricks/operators/databricks.py
@@ -29,7 +29,7 @@
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
-from airflow.models import BaseOperator, BaseOperatorLink, XCom
+from airflow.models import BaseOperator, XCom
 from airflow.providers.databricks.hooks.databricks import DatabricksHook, RunLifeCycleState, RunState
 from airflow.providers.databricks.operators.databricks_workflow import (
     DatabricksWorkflowTaskGroup,
@@ -41,12 +41,18 @@
 )
 from airflow.providers.databricks.triggers.databricks import DatabricksExecutionTrigger
 from airflow.providers.databricks.utils.databricks import normalise_json_content, validate_trigger_event
+from airflow.providers.databricks.version_compat import AIRFLOW_V_3_0_PLUS
 
 if TYPE_CHECKING:
     from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.utils.context import Context
     from airflow.utils.task_group import TaskGroup
 
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import BaseOperatorLink
+else:
+    from airflow.models.baseoperatorlink import BaseOperatorLink  # type: ignore[no-redef]
+
 DEFER_METHOD_NAME = "execute_complete"
 XCOM_RUN_ID_KEY = "run_id"
 XCOM_JOB_ID_KEY = "job_id"

diff --git a/providers/databricks/src/airflow/providers/databricks/plugins/databricks_workflow.py b/providers/databricks/src/airflow/providers/databricks/plugins/databricks_workflow.py
@@ -27,7 +27,7 @@
 from flask_appbuilder.api import expose
 
 from airflow.exceptions import AirflowException, TaskInstanceNotFound
-from airflow.models import BaseOperator, BaseOperatorLink, DagBag
+from airflow.models import DagBag
 from airflow.models.dag import DAG, clear_task_instances
 from airflow.models.dagrun import DagRun
 from airflow.models.taskinstance import TaskInstance, TaskInstanceKey
@@ -48,8 +48,14 @@
 if TYPE_CHECKING:
     from sqlalchemy.orm.session import Session
 
+    from airflow.models import BaseOperator
     from airflow.providers.databricks.operators.databricks import DatabricksTaskBaseOperator
 
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import BaseOperatorLink
+else:
+    from airflow.models.baseoperatorlink import BaseOperatorLink  # type: ignore[no-redef]
+
 
 REPAIR_WAIT_ATTEMPTS = os.getenv("DATABRICKS_REPAIR_WAIT_ATTEMPTS", 20)
 REPAIR_WAIT_DELAY = os.getenv("DATABRICKS_REPAIR_WAIT_DELAY", 0.5)

@@ -69,15 +69,16 @@ You can install such cross-provider dependencies when installing from PyPI. For
 
 .. code-block:: bash
 
-    pip install apache-airflow-providers-dbt-cloud[http]
+    pip install apache-airflow-providers-dbt-cloud[common.compat]
 
 
-==============================================================================================================  ===============
-Dependent package                                                                                               Extra
-==============================================================================================================  ===============
-`apache-airflow-providers-http <https://airflow.apache.org/docs/apache-airflow-providers-http>`_                ``http``
-`apache-airflow-providers-openlineage <https://airflow.apache.org/docs/apache-airflow-providers-openlineage>`_  ``openlineage``
-==============================================================================================================  ===============
+==================================================================================================================  =================
+Dependent package                                                                                                   Extra
+==================================================================================================================  =================
+`apache-airflow-providers-common-compat <https://airflow.apache.org/docs/apache-airflow-providers-common-compat>`_  ``common.compat``
+`apache-airflow-providers-http <https://airflow.apache.org/docs/apache-airflow-providers-http>`_                    ``http``
+`apache-airflow-providers-openlineage <https://airflow.apache.org/docs/apache-airflow-providers-openlineage>`_      ``openlineage``
+==================================================================================================================  =================
 
 The changelog for the provider package can be found in the
 `changelog <https://airflow.apache.org/docs/apache-airflow-providers-dbt-cloud/4.2.0/changelog.html>`_.
@@ -70,12 +70,16 @@ dependencies = [
 "openlineage" = [
     "apache-airflow-providers-openlineage>=1.7.0",
 ]
+"common.compat" = [
+    "apache-airflow-providers-common-compat"
+]
 
 [dependency-groups]
 dev = [
     "apache-airflow",
     "apache-airflow-task-sdk",
     "apache-airflow-devel-common",
+    "apache-airflow-providers-common-compat",
     "apache-airflow-providers-http",
     "apache-airflow-providers-openlineage",
     # Additional devel dependencies (do not remove this line and add extra development dependencies)

@@ -98,6 +98,9 @@ def get_provider_info():
             "asgiref>=2.3.0",
             "aiohttp>=3.9.2",
         ],
-        "optional-dependencies": {"openlineage": ["apache-airflow-providers-openlineage>=1.7.0"]},
+        "optional-dependencies": {
+            "openlineage": ["apache-airflow-providers-openlineage>=1.7.0"],
+            "common.compat": ["apache-airflow-providers-common-compat"],
+        },
         "devel-dependencies": [],
     }
@@ -24,7 +24,8 @@
 from typing import TYPE_CHECKING, Any
 
 from airflow.configuration import conf
-from airflow.models import BaseOperator, BaseOperatorLink, XCom
+from airflow.models import BaseOperator, XCom
+from airflow.providers.common.compat.version_compat import AIRFLOW_V_3_0_PLUS
 from airflow.providers.dbt.cloud.hooks.dbt import (
     DbtCloudHook,
     DbtCloudJobRunException,
@@ -38,6 +39,11 @@
     from airflow.providers.openlineage.extractors import OperatorLineage
     from airflow.utils.context import Context
 
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import BaseOperatorLink
+else:
+    from airflow.models.baseoperatorlink import BaseOperatorLink  # type: ignore[no-redef]
+
 
 class DbtCloudRunJobOperatorLink(BaseOperatorLink):
     """Allows users to monitor the triggered job run directly in dbt Cloud."""

@@ -19,12 +19,17 @@
 
 from typing import TYPE_CHECKING, ClassVar
 
-from airflow.models import BaseOperatorLink, XCom
+from airflow.models import XCom
+from airflow.providers.google.version_compat import AIRFLOW_V_3_0_PLUS
 
 if TYPE_CHECKING:
     from airflow.models import BaseOperator
     from airflow.models.taskinstancekey import TaskInstanceKey
 
+if AIRFLOW_V_3_0_PLUS:
+    from airflow.sdk import BaseOperatorLink
+else:
+    from airflow.models.baseoperatorlink import BaseOperatorLink  # type: ignore[no-redef]
 
 BASE_LINK = "https://console.cloud.google.com"