Rework ProvidersManager to separate runtime and infrastructure focus

.pre-commit-config.yaml

@@ -596,6 +596,7 @@ repos:
           ^airflow-core/src/airflow/utils/db\.py$|
           ^airflow-core/src/airflow/utils/trigger_rule\.py$|
           ^airflow-core/tests/|
+          ^task-sdk/tests/|
           ^.*changelog\.(rst|txt)$|
           ^.*CHANGELOG\.(rst|txt)$|
           ^chart/values.schema\.json$|

airflow-core/pyproject.toml

@@ -240,6 +240,7 @@ exclude = [
 "../shared/timezones/src/airflow_shared/timezones" = "src/airflow/_shared/timezones"
 "../shared/listeners/src/airflow_shared/listeners" = "src/airflow/_shared/listeners"
 "../shared/plugins_manager/src/airflow_shared/plugins_manager" = "src/airflow/_shared/plugins_manager"
+"../shared/providers_discovery/src/airflow_shared/providers_discovery" = "src/airflow/_shared/providers_discovery"
 
 [tool.hatch.build.targets.custom]
 path = "./hatch_build.py"
@@ -317,4 +318,5 @@ shared_distributions = [
     "apache-airflow-shared-secrets-masker",
     "apache-airflow-shared-timezones",
     "apache-airflow-shared-plugins-manager",
+    "apache-airflow-shared-providers-discovery",
 ]

airflow-core/src/airflow/_shared/providers_discovery

@@ -0,0 +1 @@
+../../../../shared/providers_discovery/src/airflow_shared/providers_discovery

airflow-core/src/airflow/providers_manager.py

@@ -33,9 +33,9 @@
 from time import perf_counter
 from typing import TYPE_CHECKING, Any, NamedTuple, ParamSpec, TypeVar, cast
 
-from packaging.utils import canonicalize_name
-
-from airflow._shared.module_loading import entry_points_with_dist, import_string
+from airflow import DeprecatedImportWarning
+from airflow._shared.module_loading import import_string
+from airflow._shared.providers_discovery import discover_all_providers_from_packages
 from airflow.exceptions import AirflowOptionalProviderFeatureException
 from airflow.utils.log.logging_mixin import LoggingMixin
 
@@ -438,6 +438,33 @@ def __init__(self):
         self._plugins_set: set[PluginInfo] = set()
         self._init_airflow_core_hooks()
 
+        self._runtime_manager = None
+
+    def __getattribute__(self, name: str):
+        # Hacky but does the trick for now
+        runtime_properties = {
+            "hooks",
+            "taskflow_decorators",
+            "filesystem_module_names",
+            "asset_factories",
+            "asset_uri_handlers",
+            "asset_to_openlineage_converters",
+        }
+
+        if name in runtime_properties:
+            warnings.warn(
+                f"ProvidersManager.{name} is deprecated. Use ProvidersManagerTaskRuntime.{name} from task-sdk instead.",
+                DeprecatedImportWarning,
+                stacklevel=2,
+            )
+            if object.__getattribute__(self, "_runtime_manager") is None:
+                from airflow.sdk.providers_manager_runtime import ProvidersManagerTaskRuntime
+
+                object.__setattr__(self, "_runtime_manager", ProvidersManagerTaskRuntime())
+            return getattr(object.__getattribute__(self, "_runtime_manager"), name)
+
+        return object.__getattribute__(self, name)
+
     def _init_airflow_core_hooks(self):
         """Initialize the hooks dict with default hooks from Airflow core."""
         core_dummy_hooks = {
@@ -472,7 +499,7 @@ def initialize_providers_list(self):
         # Development purpose. In production provider.yaml files are not present in the 'airflow" directory
         # So there is no risk we are going to override package provider accidentally. This can only happen
         # in case of local development
-        self._discover_all_providers_from_packages()
+        discover_all_providers_from_packages(self._provider_dict, self._provider_schema_validator)
         self._verify_all_providers_all_compatible()
         self._provider_dict = dict(sorted(self._provider_dict.items()))
 
@@ -607,57 +634,6 @@ def initialize_providers_cli_command(self):
         self.initialize_providers_list()
         self._discover_cli_command()
 
-    def _discover_all_providers_from_packages(self) -> None:
-        """
-        Discover all providers by scanning packages installed.
-
-        The list of providers should be returned via the 'apache_airflow_provider'
-        entrypoint as a dictionary conforming to the 'airflow/provider_info.schema.json'
-        schema. Note that the schema is different at runtime than provider.yaml.schema.json.
-        The development version of provider schema is more strict and changes together with
-        the code. The runtime version is more relaxed (allows for additional properties)
-        and verifies only the subset of fields that are needed at runtime.
-        """
-        for entry_point, dist in entry_points_with_dist("apache_airflow_provider"):
-            if not dist.metadata:
-                continue
-            package_name = canonicalize_name(dist.metadata["name"])
-            if package_name in self._provider_dict:
-                continue
-            log.debug("Loading %s from package %s", entry_point, package_name)
-            version = dist.version
-            provider_info = entry_point.load()()
-            self._provider_schema_validator.validate(provider_info)
-            provider_info_package_name = provider_info["package-name"]
-            if package_name != provider_info_package_name:
-                raise ValueError(
-                    f"The package '{package_name}' from packaging information "
-                    f"{provider_info_package_name} do not match. Please make sure they are aligned"
-                )
-
-            # issue-59576: Retrieve the project.urls.documentation from dist.metadata
-            project_urls = dist.metadata.get_all("Project-URL")
-            documentation_url: str | None = None
-
-            if project_urls:
-                for entry in project_urls:
-                    if "," in entry:
-                        name, url = entry.split(",")
-                        if name.strip().lower() == "documentation":
-                            documentation_url = url
-                            break
-
-            provider_info["documentation-url"] = documentation_url
-
-            if package_name not in self._provider_dict:
-                self._provider_dict[package_name] = ProviderInfo(version, provider_info)
-            else:
-                log.warning(
-                    "The provider for package '%s' could not be registered from because providers for that "
-                    "package name have already been registered",
-                    package_name,
-                )
-
     def _discover_hooks_from_connection_types(
         self,
         hook_class_names_registered: set[str],