Locate Airflow provider info using the provider_info entry point

docs/source/user_guide/pipeline-components.md

@@ -461,6 +461,7 @@ Examples:
 The [Apache Airflow provider package catalog connector](https://github.com/elyra-ai/elyra/tree/main/elyra/pipeline/airflow/provider_package_catalog_connector) provides access to operators that are stored in [Apache Airflow provider packages](https://airflow.apache.org/docs/apache-airflow-providers/):
 - Only the [wheel distribution format](https://packaging.python.org/en/latest/glossary/#term-Wheel) is supported.
 - Only Airflow < 2 and operators for Airflow < 2 are supported. Use of that functionality is not working in Elyra >=4, which is no longer supporting Airflow 1.x. 
+- The provider info, which identifies the modules that contain operators, is located using the `provider_info` entry point in the `apache_airflow_provider` group that the provider package declares, as described in the [Apache Airflow documentation](https://airflow.apache.org/docs/apache-airflow-providers/howto/create-custom-providers.html). Support for provider packages that do not declare this entry point and define the provider info in a `get_provider_info.py` module is deprecated.
 - The specified URL must be retrievable using an HTTP `GET` request. `http`, `https`, and `file` [URI schemes](https://www.iana.org/assignments/uri-schemes/uri-schemes.xhtml) are supported.
 - In secured environments where SSL server authenticity can only be validated using certificates based on private public key infrastructure (PKI) with root and optionally intermediate certificate authorities (CAs) that are not publicly trusted, you must define environment variable `TRUSTED_CA_BUNDLE_PATH` in the environment where JupyterLab/Elyra is running. The variable value must identify an existing [Privacy-Enhanced Mail (PEM) file](https://en.wikipedia.org/wiki/Privacy-Enhanced_Mail).
 

elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py

@@ -15,6 +15,7 @@
 #
 
 import ast
+from importlib.metadata import Distribution
 import os
 from os.path import sep
 from pathlib import Path
@@ -25,6 +26,7 @@
 from typing import Dict
 from typing import List
 from typing import Optional
+from typing import Tuple
 from urllib.parse import urlparse
 import zipfile
 
@@ -158,35 +160,46 @@ def get_catalog_entries(self, catalog_metadata: Dict[str, Any]) -> List[Dict[str
             self.log.debug(f"Deleting downloaded Airflow provider archive '{archive}' ...")
             os.remove(archive)
 
-            # Locate 'get_provider_info.py' in the extracted archive.
-            # It identifies Python modules that contain operator class definitions.
-            pl = list(self.tmp_archive_dir.glob("**/get_provider_info.py"))
-            if len(pl) != 1:
-                # No such file or more than one file was found. Cannot proceed.
-                self.log.error(
-                    f"Error. Airflow provider package connector '{catalog_metadata.get('display_name')}' "
-                    f"is not configured properly. The archive '{archive}' "
-                    f"contains {len(pl)} file(s) named 'get_provider_info.py'."
-                )
+            # Locate the function that returns the provider info.
+            # The provider info identifies Python modules that contain operator class definitions.
+            provider_info_location = self._locate_provider_info(catalog_metadata, archive)
+            if provider_info_location is None:
                 # return empty list
                 return operator_key_list
-            get_provider_info_file_location = pl[0]
+            provider_info_file_location, provider_info_function = provider_info_location
 
-            # extract module names from get_provider_info.py
-            with open(get_provider_info_file_location, "r") as gpi_file:
+            # extract module names from the provider info
+            with open(provider_info_file_location, "r") as gpi_file:
                 f = gpi_file.read()
-            namespace = {}
-            exec(f, namespace)
+            # The module is run without its package context, because the provider package is not
+            # installed. Its name is not qualified, so relative imports raise an ImportError, too.
+            namespace = {"__name__": "provider_info"}
+            try:
+                exec(f, namespace)
+            except ImportError as ex:
+                self.log.error(
+                    f"Error. Airflow provider package connector '{catalog_metadata.get('display_name')}' "
+                    f"cannot load the provider info module '{provider_info_file_location}': {ex}. "
+                    "The module must not import from its own package or from other packages "
+                    "that are not installed."
+                )
+                return operator_key_list
             try:
-                # try to run the 'get_provider_info' method
-                return_dict = namespace["get_provider_info"]()
-            except KeyError:
-                # no method with this name is defined in get_provider_info.py
+                # resolve the provider info function, which might be an attribute of
+                # an object in the module, e.g. 'ProviderInfo.get'
+                attrs = provider_info_function.split(".")
+                provider_info = namespace[attrs[0]]
+                for attr in attrs[1:]:
+                    provider_info = getattr(provider_info, attr)
+            except (KeyError, AttributeError):
+                # no function with this name is defined in the provider info module
                 self.log.error(
                     f"Error. Airflow provider package connector '{catalog_metadata.get('display_name')}' "
-                    f"cannot invoke ' get_provider_info' method in '{get_provider_info_file_location}'."
+                    f"cannot invoke '{provider_info_function}' method in '{provider_info_file_location}'."
                 )
                 return operator_key_list
+            # run the provider info function
+            return_dict = provider_info()
 
             python_scripts = []
             for operator_entry in return_dict.get("operators", []):
@@ -327,6 +340,70 @@ def get_catalog_entries(self, catalog_metadata: Dict[str, Any]) -> List[Dict[str
 
         return operator_key_list
 
+    def _locate_provider_info(self, catalog_metadata: Dict[str, Any], archive: str) -> Optional[Tuple[Path, str]]:
+        """
+        Locate the function that returns the provider info in the extracted provider package archive.
+
+        Provider packages declare this function as the 'provider_info' entry point in the
+        'apache_airflow_provider' group, see
+        https://airflow.apache.org/docs/apache-airflow-providers/howto/create-custom-providers.html
+        If no such entry point is declared, fall back to the 'get_provider_info' function
+        in the 'get_provider_info.py' module. This fallback is deprecated.
+
+        :param catalog_metadata: Contains the catalog connector configuration information.
+        :param archive: The name of the provider package archive, which is used in error messages.
+        :returns: The location of the module that defines the function and the function name,
+                  or None if the function cannot be located
+        """
+        entry_point = next(
+            (
+                entry_point
+                for dist_info_dir in self.tmp_archive_dir.glob("*.dist-info")
+                for entry_point in Distribution.at(dist_info_dir).entry_points.select(
+                    group="apache_airflow_provider", name="provider_info"
+                )
+            ),
+            None,
+        )
+        if entry_point:
+            if not entry_point.attr:
+                self.log.error(
+                    f"Error. Airflow provider package connector '{catalog_metadata.get('display_name')}' "
+                    f"is not configured properly. The 'provider_info' entry point '{entry_point.value}' "
+                    f"of the archive '{archive}' does not reference a function."
+                )
+                return None
+            module_path = self.tmp_archive_dir.joinpath(*entry_point.module.split("."))
+            # the module is either a Python script or a package
+            for module_file in [module_path.with_suffix(".py"), module_path / "__init__.py"]:
+                if module_file.is_file():
+                    return module_file, entry_point.attr
+            self.log.error(
+                f"Error. Airflow provider package connector '{catalog_metadata.get('display_name')}' "
+                f"is not configured properly. The archive '{archive}' "
+                f"does not contain the module '{entry_point.module}' of the 'provider_info' entry point."
+            )
+            return None
+
+        # Locate 'get_provider_info.py' in the extracted archive.
+        pl = list(self.tmp_archive_dir.glob("**/get_provider_info.py"))
+        if len(pl) != 1:
+            # No such file or more than one file was found. Cannot proceed.
+            self.log.error(
+                f"Error. Airflow provider package connector '{catalog_metadata.get('display_name')}' "
+                f"is not configured properly. The archive '{archive}' "
+                f"does not declare a 'provider_info' entry point and "
+                f"contains {len(pl)} file(s) named 'get_provider_info.py'."
+            )
+            return None
+        self.log.warning(
+            f"Warning. Airflow provider package connector '{catalog_metadata.get('display_name')}' "
+            f"located the provider info in '{pl[0].relative_to(self.tmp_archive_dir)}' because the archive "
+            f"'{archive}' does not declare a 'provider_info' entry point in the 'apache_airflow_provider' group. "
+            "Support for provider packages without this entry point is deprecated."
+        )
+        return pl[0], "get_provider_info"
+
     def get_entry_data(
         self, catalog_entry_data: Dict[str, Any], catalog_metadata: Dict[str, Any]
     ) -> Optional[EntryData]:

elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py

@@ -20,6 +20,7 @@
 from urllib.parse import urlparse
 import zipfile
 
+import pytest
 from requests import get
 
 from elyra.pipeline.airflow.provider_package_catalog_connector.airflow_provider_package_catalog_connector import (
@@ -117,6 +118,188 @@ def test_invalid_input_get_catalog_entries():
     assert len(ce) == 0
 
 
+OPERATOR_SOURCE = """
+from airflow.models import BaseOperator
+
+
+class MyOperator(BaseOperator):
+    def __init__(self, my_property: str, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+"""
+
+
+def create_provider_package(directory: Path, files: dict) -> str:
+    """
+    Create a provider package archive that contains the specified files
+    and return its URL
+    """
+    archive = directory / "my_provider-1.0.0-py3-none-any.whl"
+    with zipfile.ZipFile(archive, "w") as zip_file:
+        for name, content in files.items():
+            zip_file.writestr(name, content)
+    return archive.as_uri()
+
+
+@pytest.mark.parametrize(
+    "provider_info_files, deprecated",
+    [
+        # provider info function declared as entry point, in a module with a custom name
+        (
+            {
+                "my_provider/provider.py": "def provider_info():\n    return PROVIDER_INFO\n",
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.provider:provider_info\n"
+                ),
+            },
+            False,
+        ),
+        # provider info function declared as entry point, in a package
+        (
+            {
+                "my_provider/info/__init__.py": "def get_info():\n    return PROVIDER_INFO\n",
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[console_scripts]\nmy-script = my_provider.cli:main\n\n"
+                    "[apache_airflow_provider]\nprovider_info = my_provider.info:get_info\n"
+                ),
+            },
+            False,
+        ),
+        # provider info function declared as entry point, as a method of a class
+        (
+            {
+                "my_provider/provider.py": (
+                    "class ProviderInfo:\n    @staticmethod\n    def get():\n        return PROVIDER_INFO\n"
+                ),
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.provider:ProviderInfo.get\n"
+                ),
+            },
+            False,
+        ),
+        # no entry point, 'get_provider_info.py' module
+        (
+            {
+                "my_provider/get_provider_info.py": "def get_provider_info():\n    return PROVIDER_INFO\n",
+                "my_provider-1.0.0.dist-info/METADATA": "Name: my-provider\n",
+            },
+            True,
+        ),
+    ],
+)
+def test_locate_provider_info(caplog, provider_info_files, deprecated):
+    """
+    Test that the provider info is located using the 'provider_info' entry point, if
+    the provider package declares one, and the deprecated 'get_provider_info.py' module otherwise.
+    """
+    provider_info = {"operators": [{"python-modules": ["my_provider.operators.my_operator"]}]}
+    files = {
+        name: f"PROVIDER_INFO = {provider_info}\n\n\n{content}" if name.endswith(".py") else content
+        for name, content in provider_info_files.items()
+    }
+    files["my_provider/operators/my_operator.py"] = OPERATOR_SOURCE
+
+    with TemporaryDirectory() as dirpath:
+        url = create_provider_package(Path(dirpath), files)
+
+        appc = AirflowProviderPackageCatalogConnector(AIRFLOW_SUPPORTED_FILE_TYPES)
+        ces = appc.get_catalog_entries({"airflow_provider_package_download_url": url, "display_name": "my-provider"})
+        assert ces == [
+            {
+                "provider_package": "my_provider-1.0.0-py3-none-any.whl",
+                "provider": "my_provider",
+                "file": "my_provider/operators/my_operator.py",
+            }
+        ]
+        assert ("Support for provider packages without this entry point is deprecated." in caplog.text) is deprecated
+
+
+@pytest.mark.parametrize(
+    "files, error",
+    [
+        # the entry point module isn't in the archive
+        (
+            {
+                "my_provider/get_provider_info.py": "def get_provider_info():\n    return {}\n",
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.missing:provider_info\n"
+                ),
+            },
+            "does not contain the module 'my_provider.missing' of the 'provider_info' entry point",
+        ),
+        # the entry point function isn't defined
+        (
+            {
+                "my_provider/provider.py": "def other_function():\n    return {}\n",
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.provider:provider_info\n"
+                ),
+            },
+            "cannot invoke 'provider_info' method",
+        ),
+        # the entry point method isn't defined
+        (
+            {
+                "my_provider/provider.py": "class ProviderInfo:\n    pass\n",
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.provider:ProviderInfo.get\n"
+                ),
+            },
+            "cannot invoke 'ProviderInfo.get' method",
+        ),
+        # the entry point doesn't reference a function
+        (
+            {
+                "my_provider/provider.py": "def provider_info():\n    return {}\n",
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.provider\n"
+                ),
+            },
+            "does not reference a function",
+        ),
+        # the entry point module imports from its own package, using an absolute import
+        (
+            {
+                "my_provider/version.py": "VERSION = '1.0.0'\n",
+                "my_provider/info/__init__.py": (
+                    "from my_provider.version import VERSION\n\n\ndef get_info():\n    return {}\n"
+                ),
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.info:get_info\n"
+                ),
+            },
+            "The module must not import from its own package",
+        ),
+        # the entry point module imports from its own package, using a relative import
+        (
+            {
+                "my_provider/info/version.py": "VERSION = '1.0.0'\n",
+                "my_provider/info/__init__.py": "from .version import VERSION\n\n\ndef get_info():\n    return {}\n",
+                "my_provider-1.0.0.dist-info/entry_points.txt": (
+                    "[apache_airflow_provider]\nprovider_info = my_provider.info:get_info\n"
+                ),
+            },
+            "The module must not import from its own package",
+        ),
+        # no entry point and no 'get_provider_info.py' module
+        (
+            {"my_provider/provider.py": "def provider_info():\n    return {}\n"},
+            "does not declare a 'provider_info' entry point and contains 0 file(s) named 'get_provider_info.py'",
+        ),
+    ],
+)
+def test_locate_provider_info_errors(caplog, files, error):
+    """
+    Test that an error is logged if the provider info cannot be located
+    """
+    with TemporaryDirectory() as dirpath:
+        url = create_provider_package(Path(dirpath), files)
+
+        appc = AirflowProviderPackageCatalogConnector(AIRFLOW_SUPPORTED_FILE_TYPES)
+        ces = appc.get_catalog_entries({"airflow_provider_package_download_url": url, "display_name": "my-provider"})
+        assert ces == []
+        assert error in caplog.text
+
+
 # -----------------------------------
 # Long running test(s)
 # ----------------------------------