Move BaseOperatorLink into the separate module

.pre-commit-config.yaml

@@ -576,9 +576,24 @@ repos:
         files: ^airflow/models/dag\.py$|^airflow/(?:decorators|utils)/task_group\.py$
       - id: check-base-operator-usage
         language: pygrep
-        name: Check BaseOperator[Link] core imports
-        description: Make sure BaseOperator[Link] is imported from airflow.models.baseoperator in core
-        entry: "from airflow\\.models import.* BaseOperator"
+        name: Check BaseOperator core imports
+        description: Make sure BaseOperator is imported from airflow.models.baseoperator in core
+        entry: "from airflow\\.models import.* BaseOperator\\b"
+        files: \.py$
+        pass_filenames: true
+        exclude: >
+          (?x)
+          ^airflow/decorators/.*$|
+          ^airflow/hooks/.*$|
+          ^airflow/operators/.*$|
+          ^airflow/providers/.*$|
+          ^airflow/sensors/.*$|
+          ^dev/provider_packages/.*$
+      - 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"
         files: \.py$
         pass_filenames: true
         exclude: >
@@ -593,7 +608,7 @@ repos:
         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 import.* BaseOperator"
+        entry: "from airflow\\.models\\.baseoperator(link)? import.* BaseOperator"
         pass_filenames: true
         files: >
           (?x)

STATIC_CODE_CHECKS.rst

@@ -154,7 +154,8 @@ require Breeze Docker image to be built locally.
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
 | check-base-operator-partial-arguments                     | Check BaseOperator and partial() arguments                   |         |
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
-| check-base-operator-usage                                 | * Check BaseOperator[Link] core imports                      |         |
+| check-base-operator-usage                                 | * Check BaseOperator core imports                            |         |
+|                                                           | * Check BaseOperatorLink core imports                        |         |
 |                                                           | * Check BaseOperator[Link] other imports                     |         |
 +-----------------------------------------------------------+--------------------------------------------------------------+---------+
 | check-boring-cyborg-configuration                         | Checks for Boring Cyborg configuration consistency           |         |

airflow/models/__init__.py

@@ -84,7 +84,7 @@ def __getattr__(name):
     "ID_LEN": "airflow.models.base",
     "Base": "airflow.models.base",
     "BaseOperator": "airflow.models.baseoperator",
-    "BaseOperatorLink": "airflow.models.baseoperator",
+    "BaseOperatorLink": "airflow.models.baseoperatorlink",
     "Connection": "airflow.models.connection",
     "DagBag": "airflow.models.dagbag",
     "DagModel": "airflow.models.dag",
@@ -115,7 +115,8 @@ def __getattr__(name):
     # I was unable to get mypy to respect a airflow/models/__init__.pyi, so
     # having to resort back to this hacky method
     from airflow.models.base import ID_LEN, Base
-    from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+    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

airflow/models/abstractoperator.py

@@ -48,7 +48,8 @@
     import jinja2  # Slow import.
     from sqlalchemy.orm import Session
 
-    from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+    from airflow.models.baseoperator import BaseOperator
+    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.dag import DAG
     from airflow.models.mappedoperator import MappedOperator
     from airflow.models.operator import Operator

airflow/models/baseoperator.py

@@ -26,15 +26,13 @@
 import logging
 import sys
 import warnings
-from abc import ABCMeta, abstractmethod
 from datetime import datetime, timedelta
 from inspect import signature
 from types import FunctionType
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
-    ClassVar,
     Collection,
     Iterable,
     Sequence,
@@ -101,12 +99,10 @@
     import jinja2  # Slow import.
     from sqlalchemy.orm import Session
 
-    from airflow.models.abstractoperator import (
-        TaskStateChangeCallback,
-    )
+    from airflow.models.abstractoperator import TaskStateChangeCallback
+    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.dag import DAG
     from airflow.models.operator import Operator
-    from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.models.xcom_arg import XComArg
     from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
     from airflow.triggers.base import BaseTrigger
@@ -1896,31 +1892,30 @@ def chain_linear(*elements: DependencyMixin | Sequence[DependencyMixin]):
         raise ValueError("No dependencies were set. Did you forget to expand with `*`?")
 
 
-@attr.s(auto_attribs=True)
-class BaseOperatorLink(metaclass=ABCMeta):
-    """Abstract base class that defines how we get an operator link."""
-
-    operators: ClassVar[list[type[BaseOperator]]] = []
+def __getattr__(name):
     """
-    This property will be used by Airflow Plugins to find the Operators to which you want
-    to assign this Operator Link
+    PEP-562: Lazy loaded attributes on python modules.
 
-    :return: List of Operator classes used by task for which you want to create extra link
+    :meta private:
     """
+    path = __deprecated_imports.get(name)
+    if not path:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
-    @property
-    @abstractmethod
-    def name(self) -> str:
-        """Name of the link. This will be the button name on the task UI."""
+    from airflow.utils.module_loading import import_string
 
-    @abstractmethod
-    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
-        """Link to external system.
+    warnings.warn(
+        f"Import `{__name__}.{name}` is deprecated. Please use `{path}.{name}`.",
+        RemovedInAirflow3Warning,
+        stacklevel=2,
+    )
+    val = import_string(f"{path}.{name}")
 
-        Note: The old signature of this function was ``(self, operator, dttm: datetime)``. That is still
-        supported at runtime but is deprecated.
+    # Store for next time
+    globals()[name] = val
+    return val
 
-        :param operator: The Airflow operator object this link is associated to.
-        :param ti_key: TaskInstance ID to return link for.
-        :return: link to external system
-        """
+
+__deprecated_imports = {
+    "BaseOperatorLink": "airflow.models.baseoperatorlink",
+}

airflow/models/baseoperatorlink.py

@@ -0,0 +1,57 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from abc import ABCMeta, abstractmethod
+from typing import TYPE_CHECKING, ClassVar
+
+import attr
+
+if TYPE_CHECKING:
+    from airflow.models.baseoperator import BaseOperator
+    from airflow.models.taskinstancekey import TaskInstanceKey
+
+
+@attr.s(auto_attribs=True)
+class BaseOperatorLink(metaclass=ABCMeta):
+    """Abstract base class that defines how we get an operator link."""
+
+    operators: ClassVar[list[type[BaseOperator]]] = []
+    """
+    This property will be used by Airflow Plugins to find the Operators to which you want
+    to assign this Operator Link
+
+    :return: List of Operator classes used by task for which you want to create extra link
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Name of the link. This will be the button name on the task UI."""
+
+    @abstractmethod
+    def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
+        """Link to external system.
+
+        Note: The old signature of this function was ``(self, operator, dttm: datetime)``. That is still
+        supported at runtime but is deprecated.
+
+        :param operator: The Airflow operator object this link is associated to.
+        :param ti_key: TaskInstance ID to return link for.
+        :return: link to external system
+        """

airflow/models/mappedoperator.py

@@ -66,7 +66,8 @@
     from airflow.models.abstractoperator import (
         TaskStateChangeCallback,
     )
-    from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+    from airflow.models.baseoperator import BaseOperator
+    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.dag import DAG
     from airflow.models.expandinput import (
         ExpandInput,

airflow/operators/trigger_dagrun.py

@@ -28,7 +28,8 @@
 from airflow.api.common.trigger_dag import trigger_dag
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, DagNotFound, DagRunAlreadyExists
-from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+from airflow.models.baseoperator import BaseOperator
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.models.dag import DagModel
 from airflow.models.dagbag import DagBag
 from airflow.models.dagrun import DagRun

airflow/sensors/external_task.py

@@ -27,7 +27,7 @@
 
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException, AirflowSkipException, RemovedInAirflow3Warning
-from airflow.models.baseoperator import BaseOperatorLink
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.models.dag import DagModel
 from airflow.models.dagbag import DagBag
 from airflow.models.dagrun import DagRun

airflow/serialization/serialized_objects.py

@@ -70,7 +70,7 @@
 
     from pydantic import BaseModel
 
-    from airflow.models.baseoperator import BaseOperatorLink
+    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.expandinput import ExpandInput
     from airflow.models.operator import Operator
     from airflow.models.taskmixin import DAGNode

docs/apache-airflow/howto/define-extra-link.rst

@@ -27,7 +27,8 @@ The following code shows how to add extra links to an operator via Plugins:
 
 .. code-block:: python
 
-    from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+    from airflow.models.baseoperator import BaseOperator
+    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.plugins_manager import AirflowPlugin
 
@@ -80,7 +81,8 @@ tasks using :class:`~airflow.providers.amazon.aws.transfers.gcs_to_s3.GCSToS3Ope
 
 .. code-block:: python
 
-  from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+  from airflow.models.baseoperator import BaseOperator
+  from airflow.models.baseoperatorlink 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
@@ -118,7 +120,8 @@ Console, but if we wanted to change that link we could:
 
 .. code-block:: python
 
-    from airflow.models.baseoperator import BaseOperatorLink, BaseOperator
+    from airflow.models.baseoperator import BaseOperator
+    from airflow.models.baseoperatorlink import BaseOperatorLink
     from airflow.models.taskinstancekey import TaskInstanceKey
     from airflow.models.xcom import XCom
     from airflow.plugins_manager import AirflowPlugin

docs/apache-airflow/public-airflow-interface.rst

@@ -312,6 +312,13 @@ 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

docs/conf.py

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

pyproject.toml

@@ -149,6 +149,7 @@ combine-as-imports = true
 [tool.ruff.flake8-tidy-imports.banned-api]
 "airflow.AirflowException".msg = "Use airflow.exceptions.AirflowException instead."
 "airflow.Dataset".msg = "Use airflow.datasets.Dataset instead."
+"airflow.models.baseoperator.BaseOperatorLink".msg = "Use airflow.models.baseoperatorlink.BaseOperatorLink"
 
 [tool.ruff.flake8-tidy-imports]
 # Ban certain modules from being imported at module level, instead requiring

tests/api_connexion/endpoints/test_extra_link_endpoint.py

@@ -22,7 +22,7 @@
 import pytest
 
 from airflow.api_connexion.exceptions import EXCEPTIONS_LINK_MAP
-from airflow.models.baseoperator import BaseOperatorLink
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.models.dag import DAG
 from airflow.models.dagbag import DagBag
 from airflow.models.xcom import XCom

tests/api_connexion/endpoints/test_plugin_endpoint.py

@@ -23,7 +23,7 @@
 from flask_appbuilder import BaseView
 
 from airflow.hooks.base import BaseHook
-from airflow.models.baseoperator import BaseOperatorLink
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.plugins_manager import AirflowPlugin
 from airflow.security import permissions
 from airflow.ti_deps.deps.base_ti_dep import BaseTIDep

tests/api_connexion/schemas/test_plugin_schema.py

@@ -25,7 +25,7 @@
     plugin_schema,
 )
 from airflow.hooks.base import BaseHook
-from airflow.models.baseoperator import BaseOperatorLink
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.plugins_manager import AirflowPlugin
 
 

tests/serialization/test_dag_serialization.py

@@ -43,7 +43,8 @@
 from airflow.decorators.base import DecoratedOperator
 from airflow.exceptions import AirflowException, SerializationError
 from airflow.hooks.base import BaseHook
-from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+from airflow.models.baseoperator import BaseOperator
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.models.connection import Connection
 from airflow.models.dag import DAG
 from airflow.models.dagbag import DagBag

tests/test_utils/mock_operators.py

@@ -21,7 +21,8 @@
 
 import attr
 
-from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+from airflow.models.baseoperator import BaseOperator
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.models.xcom import XCom
 
 if TYPE_CHECKING:

tests/www/views/test_views_extra_links.py

@@ -23,7 +23,8 @@
 
 import pytest
 
-from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+from airflow.models.baseoperator import BaseOperator
+from airflow.models.baseoperatorlink import BaseOperatorLink
 from airflow.models.dag import DAG
 from airflow.utils import timezone
 from airflow.utils.state import DagRunState