From 2898dd66a4b3049a3d7fb4af1f58afac54841e31 Mon Sep 17 00:00:00 2001 From: NopeCharon <115794254+NopeCharon@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:26:41 +0530 Subject: [PATCH 1/2] Locate Airflow provider info using the entry point The Airflow provider package catalog connector looked for a module named 'get_provider_info.py' in the provider package to determine the modules that define operators. Airflow doesn't require this module name. Provider packages declare the function that returns the provider info as the 'provider_info' entry point in the 'apache_airflow_provider' group, and the function can be defined in any module. Packages that use another module name could not be added to a catalog. Read the entry point from the package's dist-info metadata and load the declared function from its module, which may be a Python script or a package. Packages that don't declare the entry point fall back to the 'get_provider_info.py' module, as before. Fixes #3152 Signed-off-by: NopeCharon <115794254+NopeCharon@users.noreply.github.com> --- ...flow_provider_package_catalog_connector.py | 81 +++++++++--- ...test_airflow_provider_package_connector.py | 117 ++++++++++++++++++ 2 files changed, 181 insertions(+), 17 deletions(-) diff --git a/elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py b/elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py index 39b1903cd..b746c3dd1 100644 --- a/elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py +++ b/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,33 +160,27 @@ 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) try: - # try to run the 'get_provider_info' method - return_dict = namespace["get_provider_info"]() + # try to run the provider info function + return_dict = namespace[provider_info_function]() except KeyError: - # no method with this name is defined in get_provider_info.py + # 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 @@ -327,6 +323,57 @@ 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. + + :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: + 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 + return pl[0], "get_provider_info" + def get_entry_data( self, catalog_entry_data: Dict[str, Any], catalog_metadata: Dict[str, Any] ) -> Optional[EntryData]: diff --git a/elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py b/elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py index 394477984..dc5e59c18 100644 --- a/elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py +++ b/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,122 @@ 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", + [ + # 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" + ), + }, + # 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" + ), + }, + # 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", + }, + ], +) +def test_locate_provider_info(provider_info_files): + """ + Test that the provider info is located using the 'provider_info' entry point, if + the provider package declares one, and the '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", + } + ] + + +@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", + ), + # 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) # ---------------------------------- From fbdca085dd1bd55db9ca67c5648820b330027c36 Mon Sep 17 00:00:00 2001 From: NopeCharon <115794254+NopeCharon@users.noreply.github.com> Date: Mon, 2 Nov 2026 11:07:50 +0530 Subject: [PATCH 2/2] Handle more provider info entry point cases Run the provider info module with a module name, so that relative imports raise an ImportError, and log a specific error if the module cannot be loaded because it imports from its own package or another package that isn't installed. Resolve entry point attributes that reference a method, such as 'module:Info.get', and log an error if the entry point doesn't reference a function. Log a deprecation warning if the provider info is located using the 'get_provider_info.py' module because the package doesn't declare the 'provider_info' entry point, and document how the connector locates the provider info. Signed-off-by: NopeCharon <115794254+NopeCharon@users.noreply.github.com> --- docs/source/user_guide/pipeline-components.md | 1 + ...flow_provider_package_catalog_connector.py | 42 ++++++- ...test_airflow_provider_package_connector.py | 106 ++++++++++++++---- 3 files changed, 123 insertions(+), 26 deletions(-) diff --git a/docs/source/user_guide/pipeline-components.md b/docs/source/user_guide/pipeline-components.md index 88054c5da..6336486c6 100644 --- a/docs/source/user_guide/pipeline-components.md +++ b/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). diff --git a/elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py b/elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py index b746c3dd1..ff6520250 100644 --- a/elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py +++ b/elyra/pipeline/airflow/provider_package_catalog_connector/airflow_provider_package_catalog_connector.py @@ -171,18 +171,35 @@ def get_catalog_entries(self, catalog_metadata: Dict[str, Any]) -> List[Dict[str # 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: - # try to run the provider info function - return_dict = namespace[provider_info_function]() - except KeyError: + 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: + # 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 '{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", []): @@ -331,7 +348,7 @@ def _locate_provider_info(self, catalog_metadata: Dict[str, Any], archive: str) '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. + 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. @@ -349,6 +366,13 @@ def _locate_provider_info(self, catalog_metadata: Dict[str, Any], archive: str) 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"]: @@ -372,6 +396,12 @@ def _locate_provider_info(self, catalog_metadata: Dict[str, Any], archive: str) 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( diff --git a/elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py b/elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py index dc5e59c18..e08aab7dd 100644 --- a/elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py +++ b/elyra/tests/pipeline/airflow/test_airflow_provider_package_connector.py @@ -141,34 +141,55 @@ def create_provider_package(directory: Path, files: dict) -> str: @pytest.mark.parametrize( - "provider_info_files", + "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" - ), - }, + ( + { + "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" - ), - }, + ( + { + "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", - }, + ( + { + "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(provider_info_files): +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 'get_provider_info.py' module otherwise. + 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 = { @@ -189,6 +210,7 @@ def test_locate_provider_info(provider_info_files): "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( @@ -214,6 +236,50 @@ def test_locate_provider_info(provider_info_files): }, "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"},