You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2020/11/20 16:40:02 UTC
[GitHub] [airflow] potiuk opened a new pull request #12512: Implement reading provider information from packages/sources
potiuk opened a new pull request #12512:
URL: https://github.com/apache/airflow/pull/12512
This PR implements discovering and readin provider information from
packages (using entry_points) and - if found - from local
provider yaml files for the built-in airflow providers,
when they are found in the airflow.provider packages.
The provider.yaml files - if found - take precedence over the
package-provided ones.
Add displaying provider information in CLI
Closes: #12470
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528695697
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
+ """
+ Parses found provider.yaml file and adds found provider to the list of all providers.
+ :param filename: name of the provider.yaml file
+ :param folder: folder where the provider.yaml file is
+ :param root_path: root path where we started to search ("airflow/providers" folder from the
+ local sources
+ """
+ try:
+ with open(os.path.join(folder, filename)) as provider_yaml_file:
+ provider_info = yaml.safe_load(provider_yaml_file.read())
+ self._validator.validate(provider_info)
+ package_name = "apache-airflow-providers" + folder[len(root_path) :].replace(os.sep, "-")
+ version = provider_info['versions'][0]
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
Review comment:
I want to keep it in Ordered dict so that they are returned by default sorted alphabetically. Just to make a consistent iteration and display, you never know which direction they are coming from. I want to avoid sorting every time, I could of course add it to dict and recreate the dict. Maybe that will be nicer indeed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528769479
##########
File path: dev/provider_packages/prepare_provider_packages.py
##########
@@ -1133,6 +1133,14 @@ def get_package_pip_name(provider_package_id: str, backport_packages: bool):
return f"apache-airflow-providers-{provider_package_id.replace('.', '-')}"
+def get_provider_info(provider_package_id: str):
+ provider_yaml_file_name = os.path.join(get_source_package_path(provider_package_id), "provider.yaml")
+ if not os.path.exists(provider_yaml_file_name):
+ raise Exception(f"The provider.yaml file is missing: {provider_yaml_file_name}")
+ with open(provider_yaml_file_name) as provider_file:
+ return provider_file.read()
Review comment:
```suggestion
return yaml.safe_load(provider_file)
```
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
```suggestion
def get_provider_info():
{{ PROVIDER_INFO }}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731715338
[The Workflow run](https://github.com/apache/airflow/actions/runs/377092152) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528782205
##########
File path: dev/provider_packages/prepare_provider_packages.py
##########
@@ -1133,6 +1133,14 @@ def get_package_pip_name(provider_package_id: str, backport_packages: bool):
return f"apache-airflow-providers-{provider_package_id.replace('.', '-')}"
+def get_provider_info(provider_package_id: str):
+ provider_yaml_file_name = os.path.join(get_source_package_path(provider_package_id), "provider.yaml")
+ if not os.path.exists(provider_yaml_file_name):
+ raise Exception(f"The provider.yaml file is missing: {provider_yaml_file_name}")
+ with open(provider_yaml_file_name) as provider_file:
+ return provider_file.read()
Review comment:
Seems it does even in most complex providers.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528780292
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
Review comment:
Missing the license header here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732147934
> Looking good.
>
> I'm not quite convinced by the need to make everything quite so private -- it's not really a common pattern in Airflow, and by making some things public it lets users use this class too.
I can change it to protected no problem.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732225107
> I think this is all we'd have to change to remove the `yaml.safe_load` from in the generated provieer_info.
>
> It won't be the prettiest, but we can fix that by running black on it after generating the template.
I tested this and this is what the provider info for amazon package looks okay, if horrible without black formatting.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528736934
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
Actually: bigger question: Do we ship and install the provider.yaml with the dist, or do we just use this function.
Not shipping it (if all the info is contained in here anyway) might be a nice/easy way to distinguish source vs installed provider.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528780444
##########
File path: dev/provider_packages/prepare_provider_packages.py
##########
@@ -1133,6 +1133,14 @@ def get_package_pip_name(provider_package_id: str, backport_packages: bool):
return f"apache-airflow-providers-{provider_package_id.replace('.', '-')}"
+def get_provider_info(provider_package_id: str):
+ provider_yaml_file_name = os.path.join(get_source_package_path(provider_package_id), "provider.yaml")
+ if not os.path.exists(provider_yaml_file_name):
+ raise Exception(f"The provider.yaml file is missing: {provider_yaml_file_name}")
+ with open(provider_yaml_file_name) as provider_file:
+ return provider_file.read()
Review comment:
Yeah. I was going to try that and see if it works in all cases
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r527943164
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +48,72 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = {}
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
-
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ self.__find_all_providers_from_packages()
+ self.__find_all_airflow_builtin_providers_from_local_sources()
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ def __find_all_providers_from_packages(self):
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.ContextualVersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self):
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path):
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path):
+ try:
+ with open(os.path.join(folder, filename)) as provider_yaml_file:
+ provider_info = yaml.safe_load(provider_yaml_file.read())
+ self._validator.validate(provider_info)
+ package_name = "apache-airflow-providers" + folder[len(root_path) :].replace(os.sep, "-")
+ version = provider_info['versions'][0]
+ self._provider_directory[package_name] = (version, provider_info)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading '%s/%s': %s", folder, filename, e)
@property
def providers(self):
"""Returns information about available providers."""
return self._provider_directory
+
+
+# After we move to python 3.9 + we can get rid of this and use @cache from the functools
Review comment:
Oh nice!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r527867383
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +48,64 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = {}
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
-
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
-
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
- try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
- continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self.__find_all_providers_from_packages()
+ self.__find_all_airflow_builtin_providers_from_local_sources()
+
+ def __find_all_providers_from_packages(self):
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
Review comment:
An idea from pytest: to "version" the entrypoint https://docs.pytest.org/en/stable/writing_plugins.html#making-your-plugin-installable-by-others
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732383485
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] mik-laj commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
mik-laj commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528738225
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
I agree. We can serialize this object to Python. Here is a helpful function, it is tested and works with all primitive types.
https://gist.github.com/mik-laj/3ae253d5394b34447dcbd426fd19b389#file-convert-py-L31
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732250759
[The Workflow run](https://github.com/apache/airflow/actions/runs/379268265) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-733813454
@ashb -> I would love to merge that one, so that I can rebase and finish the other two :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528702311
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
+ """
+ Parses found provider.yaml file and adds found provider to the list of all providers.
+ :param filename: name of the provider.yaml file
+ :param folder: folder where the provider.yaml file is
+ :param root_path: root path where we started to search ("airflow/providers" folder from the
+ local sources
+ """
+ try:
+ with open(os.path.join(folder, filename)) as provider_yaml_file:
+ provider_info = yaml.safe_load(provider_yaml_file.read())
+ self._validator.validate(provider_info)
+ package_name = "apache-airflow-providers" + folder[len(root_path) :].replace(os.sep, "-")
+ version = provider_info['versions'][0]
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading '%s/%s': %s", folder, filename, e)
@property
def providers(self):
"""Returns information about available providers."""
return self._provider_directory
+
+
+# After we move to python 3.9 + we can get rid of this and use @cache from the functools
+cache = lru_cache(maxsize=None)
+
+
+@cache
+def get_providers_manager() -> ProvidersManager: # noqa
+ """Returns singleton instance of ProvidersManager"""
+ return ProvidersManager()
Review comment:
Surely, There is no standard way of getting singleton in Python but if that's the pattern that is already used in Airflow, no problem with that.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528786126
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
Oh, missing `return` here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528656368
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
Review comment:
```suggestion
This 'provider.yaml' scanning takes precedence over scanning packages installed
```
"because in case" -> "in case"
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
+ """
+ Parses found provider.yaml file and adds found provider to the list of all providers.
+ :param filename: name of the provider.yaml file
+ :param folder: folder where the provider.yaml file is
+ :param root_path: root path where we started to search ("airflow/providers" folder from the
+ local sources
+ """
+ try:
+ with open(os.path.join(folder, filename)) as provider_yaml_file:
+ provider_info = yaml.safe_load(provider_yaml_file.read())
Review comment:
Can pass the stream directly to pyyaml:
```suggestion
provider_info = yaml.safe_load(provider_yaml_file)
```
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
Review comment:
If we find a match we can stop searching that tree -- we have to edit the `dirnames` (2nd arg) in place using `del` or a slice assignment.
```suggestion
for folder, subfolders, files in os.walk(path):
for filename in fnmatch.filter(files, "provider.yaml"):
self.__add_provider_info_from_local_source_file(filename, folder, root_path)
subfolders[:] = []
```
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
Review comment:
A signature change please:
```suggestion
def __add_provider_info_from_local_source_file(self, filename, provider_dist_name) -> None:
```
Where `filename` is the full path to the yaml file to load, and provider_dist_name is the name to register it as.
(dist_name not package_name, as `-` is not valid in packages).
For example, called like this:
```python
for folder, _, files in os.walk(path):
for filename in fnmatch.filter(files, "provider.yaml"):
dist_name = "apache-airflow-providers" + folder[len(root_path) + 1 :].replace(os.sep, "-")
self.__add_provider_info_from_local_source_file(os.path.join(folder, filename), dist_name)
```
I think this change will make it easier to re-use this fn for plugins.
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
+ """
+ Parses found provider.yaml file and adds found provider to the list of all providers.
+ :param filename: name of the provider.yaml file
+ :param folder: folder where the provider.yaml file is
+ :param root_path: root path where we started to search ("airflow/providers" folder from the
+ local sources
+ """
+ try:
+ with open(os.path.join(folder, filename)) as provider_yaml_file:
+ provider_info = yaml.safe_load(provider_yaml_file.read())
+ self._validator.validate(provider_info)
+ package_name = "apache-airflow-providers" + folder[len(root_path) :].replace(os.sep, "-")
+ version = provider_info['versions'][0]
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
Review comment:
How about a single dict, rather than a set and a list?
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
Do need to YAML load this - could we not return a python dict directly?
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
+ """
+ Parses found provider.yaml file and adds found provider to the list of all providers.
+ :param filename: name of the provider.yaml file
+ :param folder: folder where the provider.yaml file is
+ :param root_path: root path where we started to search ("airflow/providers" folder from the
+ local sources
+ """
+ try:
+ with open(os.path.join(folder, filename)) as provider_yaml_file:
+ provider_info = yaml.safe_load(provider_yaml_file.read())
+ self._validator.validate(provider_info)
+ package_name = "apache-airflow-providers" + folder[len(root_path) :].replace(os.sep, "-")
+ version = provider_info['versions'][0]
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading '%s/%s': %s", folder, filename, e)
@property
def providers(self):
"""Returns information about available providers."""
return self._provider_directory
+
+
+# After we move to python 3.9 + we can get rid of this and use @cache from the functools
+cache = lru_cache(maxsize=None)
+
+
+@cache
+def get_providers_manager() -> ProvidersManager: # noqa
+ """Returns singleton instance of ProvidersManager"""
+ return ProvidersManager()
Review comment:
We could do this more "formally" as Daniel did in https://github.com/apache/airflow/commit/3ca11eb9b02a2c2591292fd6b76e0e98b8f22656#diff-681de8974a439f70dfa41705f5c1681ecce615fac6c4c715c1978d28d8f0da84R63
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r531578439
##########
File path: airflow/providers_manager.py
##########
@@ -36,52 +38,147 @@
def _create_validator():
+ """Creates JSON schema validator from the provider.yaml.schema.json"""
schema = json.loads(importlib_resources.read_text('airflow', 'provider.yaml.schema.json'))
cls = jsonschema.validators.validator_for(schema)
validator = cls(schema)
return validator
class ProvidersManager:
- """Manages all provider packages."""
+ """
+ Manages all provider packages. This is a Singleton class. The first time it is
+ instantiated, it discovers all available providers in installed packages and
+ local source folders (if airflow is run from sources).
+ """
+
+ _instance = None
+ resource_version = "0"
+
+ def __new__(cls):
+ if cls._instance is None:
+ cls._instance = super().__new__(cls)
+ return cls._instance
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_dict: Dict[str, Tuple[str, Dict]] = {}
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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_airflow_builtin_providers_from_local_sources()
+ self._discover_all_providers_from_packages()
+ self._sort_provider_dictionary()
+
+ def _sort_provider_dictionary(self):
+ """
+ Sort provider_dictionary using OrderedDict.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The dictionary gets sorted so that when you iterate through it, the providers are by
+ default returned in alphabetical order.
+ """
+ sorted_dict = OrderedDict()
+ for provider_name in sorted(self._provider_dict.keys()):
+ sorted_dict[provider_name] = self._provider_dict[provider_name]
+ self._provider_dict = sorted_dict
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ def _discover_all_providers_from_packages(self) -> None:
+ """
+ Discovers 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.yaml.schema.json' schema.
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ log.debug("Loading %s from package %s", entry_point, package_name)
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_dict:
+ self._provider_dict[package_name] = (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_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ try:
+ import airflow.providers
+ except ImportError:
+ log.info("You have no providers installed.")
+ return
+ try:
+ for path in airflow.providers.__path__:
+ self._add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def _add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, subdirs, files in os.walk(path, topdown=True):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ package_name = "apache-airflow-providers" + folder[len(root_path) :].replace(os.sep, "-")
+ self._add_provider_info_from_local_source_file(os.path.join(folder, filename), package_name)
+ subdirs[:] = []
Review comment:
```suggestion
# Tell `os.walk` to not recurse any further.
subdirs[:] = []
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528762100
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
> I agree. We can serialize this object to Python. Here is a helpful function, it is tested and works with all primitive types.
> https://gist.github.com/mik-laj/3ae253d5394b34447dcbd426fd19b389#file-convert-py-L31
That was my worry about generating python dict out of the metadata (we need to generate valid python code). I am not sure if it is worth the hassle.
This code is generated on the flight, nobody will really see it, it's a very small speed improvement (if at all).
We are basically trading parsing of python code with parsing of yaml code.
WDYT @mik-laj @ashb -> do you really see a benefit of doing it (knowing that we need to add new code to handle it ?)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb edited a comment on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb edited a comment on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732227408
> > I think this is all we'd have to change to remove the `yaml.safe_load` from in the generated provieer_info.
> > It won't be the prettiest, but we can fix that by running black on it after generating the template.
>
> I tested this and this is what the provider info for amazon package looks okay, if horrible without black formatting.
Formatting with black is quite straight forward:
```
+ import black
with open(get_provider_file_path, "wt") as get_provider_file:
- get_provider_file.write(get_provider_content)
+ get_provider_file.write(black.format_str(get_provider_content, mode=black.FileMode()))
```
(In black >= 0.20.8b0, that has been renamed to just `Mode`, but the version in my container was older than that.)
It now looks like:
```
def get_provider_info():
{
"package-name": "apache-airflow-providers-amazon",
"name": "Amazon",
"description": "Amazon integration (including `Amazon Web Services (AWS) <https://aws.amazon.com/>`__).\n",
"versions": ["1.0.0b2"],
"integrations": [
{
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731278698
Hey @ashb @mik-laj -> this is the "base" version of the ProvidersManager - updated to include both options:
* running directly from the sources (in which case provider information is read directly from provider.yaml file) - dev
* run-time version that reads the information from package entry_points - production
It also includes CLI that you can query the providers installed.
Reading from packages is much slower than from file, but it is bearable (~2-3 seconds on my machine). most of it is resolving and checking package dependencies, not reading the content. Also - currently - two packages (snowflake and amazon) fail to register because they have "ContextualVersionConflict" which I hope we will eventually resolve (#10854, #10854)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk edited a comment on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk edited a comment on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731781430
Hey @ashb. It's green - this one is "pure" provider discovery - without connections/extra links - it just contains the discovery and restructured provider package tests to show better what's happening.
Right now what happens:
1) for backport providers I install airflow 1.10.12 (soon 1.10.13) install all the provider.whl and provider.tgz packages and check if all classes from airlfow.providers are importable
2) for regular providers, I prepare now "airflow" wheel package and install it from there (not from sources directly) and install all the providers from packages (whl or .tgz - I run two checks), I check if all providers are importable and then I run the new `airflow providers list` command to get all the providers and see if we see the expected number of providers as output.
This does not work now because we have import version conflict problems, but once we solve it, it should work - we have 59 out of 60 now - so for now I just skip exit in the number check.
See here: https://github.com/apache/airflow/runs/1438201616?check_suite_focus=true#step:9:4436
I will continue on rebasing the others and we can discuss tomorrow how close we can get to support the cases that you have in mind (with plugins) not only the providers. I think we can find some good future-proof solutions now, I simply do not have the capacity to also implement plugins at the same time, but I think if we implement it correctly, we can have a follow up PR implementing the "plugin" case you mentioned as well, without some heavy modifications.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732943767
[The Workflow run](https://github.com/apache/airflow/actions/runs/380992173) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731588111
[The Workflow run](https://github.com/apache/airflow/actions/runs/376156528) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731781430
Hey @ashb. It's green - this one is "pure" provider discovery - without connections/extra links - it just contains the discovery and restructured provider package tests to show better what's happening.
Right now what happens:
1) for backport providers I install airflow 1.10.12 (soon 1.10.13) install all the provider.whl and provider.tgz packages and check if all classes from airlfow.providers are importable
2) for regular providers, I prepare now "airflow" wheel package and install it from there (not from sources directly) and install all the providers from packages (whl or .tgz - I run two checks), I check if all providers are importable and then I run the new `airflow providers list` command to get all the providers and see if we see the expected number of providers as output.
This does not work now because we have import version conflict problems, but once we solve it, it should work - we have 59 out of 60 now - so for now I just skip the number check.
See here: https://github.com/apache/airflow/runs/1438201616?check_suite_focus=true#step:9:4436
I will continue on rebasing the others and we can discuss tomorrow how close we can get to support the cases that you have in mind (with plugins) not only the providers. I think we can find some good future-proof solutions now, I simply do not have the capacity to also implement plugins at the same time, but I think if we implement it correctly, we can have a follow up PR implementing the "plugin" case you mentioned as well, without some heavy modifications.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528757584
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
> Actually: bigger question: Do we ship and install the provider.yaml with the dist, or do we just use this function.
>
> Not shipping it (if all the info is contained in here anyway) might be a nice/easy way to distinguish source vs installed provider.
Not shipping - absolutely. I did not intend it (it was there in the first version but when I switched to entry_points precisely for this - being able to distinguish "package" vs. "local source".
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732176897
> > Looking good.
> > I'm not quite convinced by the need to make everything quite so private -- it's not really a common pattern in Airflow, and by making some things public it lets users use this class too.
>
> I can change it to protected no problem.
Yeah `_add` etc. would make me feel happier. I agree we should start being more careful about what we make public, but `__private` seems a bit extreme is all.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528768739
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
Oh and with .pyc files we pay that parsing time only once, YAML parse we'd pay every time.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r527943917
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +48,64 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = {}
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
-
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
-
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
- try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
- continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self.__find_all_providers_from_packages()
+ self.__find_all_airflow_builtin_providers_from_local_sources()
+
+ def __find_all_providers_from_packages(self):
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
Review comment:
Actually we can do that when we need to introduce the first such version.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528763605
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
Also with YAML we could have comments and other stuff that will be retained in such generated class (in case someone actually does look there). Not that we need it now, but I can imagine it can be useful to keep it).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731628129
[The Workflow run](https://github.com/apache/airflow/actions/runs/376446566) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528761811
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
That's possible more complex than we need @mik-laj -- could we just use `{{ PROVIDER_INFO }}` if we did a `yaml.load()` in the template scripts -- i.e. `PROVIDER_INFO` would be a python dict, and the default repr is valid python code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-733822957
I hadn't been looking cos of all the failing tests. Is it ready now?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528757584
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
> Actually: bigger question: Do we ship and install the provider.yaml with the dist, or do we just use this function.
>
> Not shipping it (if all the info is contained in here anyway) might be a nice/easy way to distinguish source vs installed provider.
Not shipping - absolutely. I did not intend it (it was there in the first version but when I switched to entry_points I removed it precisely for this reason - being able to distinguish "package" vs. "local source".
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528772350
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
> Oh and with .pyc files we pay that parsing time only once, YAML parse we'd pay every time.
True. Let me try.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528754479
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
Review comment:
See https://github.com/apache/airflow/pull/12472#discussion_r528743279 (maybe rename `find` to `discover`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-733052220
[The Workflow run](https://github.com/apache/airflow/actions/runs/381358616) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528691110
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
Review comment:
Makes sense
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk edited a comment on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk edited a comment on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731278698
Hey @ashb @mik-laj -> this is the "base" version of the ProvidersManager - updated to include both options:
* running directly from the sources (in which case provider information is read directly from provider.yaml file) - dev
* run-time version that reads the information from package entry_points - production
It also includes CLI that you can query the providers installed.
Reading from packages is much slower than from file, but it is bearable (~2-3 seconds on my machine). most of it is resolving and checking package dependencies, not reading the content. Also - currently - two packages (snowflake and amazon) fail to register because they have "ContextualVersionConflict" which I hope we will eventually resolve (#10854, #10854)
You can also try it out yourself - there is a new 'providers' cli command displaying the installed providers:
Works both - in dev env (with local sources) and prod env (with packages)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r527942296
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +48,64 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = {}
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
-
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
-
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
- try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
- continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self.__find_all_providers_from_packages()
+ self.__find_all_airflow_builtin_providers_from_local_sources()
+
+ def __find_all_providers_from_packages(self):
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
Review comment:
I think it's more like "ABI" version here, than just airflow version. (I.e. it lets us have a new form of entrypoint in the future if we want.)
Not required, I was just looking for other projects that use entrypoint to see if there's any sort of convention on what to name the entrypoints. Turns out no there isn't really.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-731732215
[The Workflow run](https://github.com/apache/airflow/actions/runs/377236180) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732339317
It look like this one is good to go. Some quarantined tests failed. I am re-running it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-734315043
gentle reminder @ash :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528692057
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
Review comment:
Good point. It takes milliseconds and it is dev-only optimisation but that's easy enough to do.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732282906
I will be running some more tests but I think everything addressed @ashb/@mik-laj
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732227408
> > I think this is all we'd have to change to remove the `yaml.safe_load` from in the generated provieer_info.
> > It won't be the prettiest, but we can fix that by running black on it after generating the template.
>
> I tested this and this is what the provider info for amazon package looks okay, if horrible without black formatting.
Formatting with black is quite straight forward:
```
+ import black
with open(get_provider_file_path, "wt") as get_provider_file:
- get_provider_file.write(get_provider_content)
+ get_provider_file.write(black.format_str(get_provider_content, mode=black.FileMode()))
```
(In black >= 0.20.8b0, that has been renamed to just `Mode`, but the version in my container was older than that.)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528688489
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
+
+ def __add_provider_info_from_local_source_file(self, filename, folder, root_path) -> None:
+ """
+ Parses found provider.yaml file and adds found provider to the list of all providers.
+ :param filename: name of the provider.yaml file
+ :param folder: folder where the provider.yaml file is
+ :param root_path: root path where we started to search ("airflow/providers" folder from the
+ local sources
+ """
+ try:
+ with open(os.path.join(folder, filename)) as provider_yaml_file:
+ provider_info = yaml.safe_load(provider_yaml_file.read())
Review comment:
True
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk merged pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk merged pull request #12512:
URL: https://github.com/apache/airflow/pull/12512
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-733847119
> I hadn't been looking cos of all the failing tests. Is it ready now?
Yep. Rebased it and but last time it passed all tests.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528768401
##########
File path: dev/provider_packages/get_provider_info_TEMPLATE.py.jinja2
##########
@@ -0,0 +1,13 @@
+import yaml
+
+try:
+ import importlib.resources as importlib_resources
+except ImportError:
+ # Try backported to PY<37 `importlib_resources`.
+ import importlib_resources
+
+
+def get_provider_info():
+ return yaml.safe_load("""
+{{ PROVIDER_INFO }}
+""")
Review comment:
We don't have to worry about the validitiy of the generated code at run-time - tests should catch any errors. But I can't see how we'd get any: safe_load only supports "primitive"/native types that python can express anyway.
That YAML file is for computers -- any comments that a user might need to see should go in to docstrings for that provider.
It just feels odd to YAML parse a static string given this file is generated anyway.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r527888405
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +48,64 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = {}
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
-
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
-
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
- try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
- continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self.__find_all_providers_from_packages()
+ self.__find_all_airflow_builtin_providers_from_local_sources()
+
+ def __find_all_providers_from_packages(self):
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
Review comment:
Sure. We can add airflow2 similarly :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732382770
[The Workflow run](https://github.com/apache/airflow/actions/runs/379564885) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Backport packages$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#issuecomment-732383149
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528720130
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
Review comment:
Takes milliseconds, right until it walks in to a node_modules future in future :)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #12512: Implement reading provider information from packages/sources
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #12512:
URL: https://github.com/apache/airflow/pull/12512#discussion_r528734625
##########
File path: airflow/providers_manager.py
##########
@@ -46,42 +50,136 @@ class ProvidersManager:
"""Manages all provider packages."""
def __init__(self):
- self._provider_directory = {}
- try:
- from airflow import providers
- except ImportError as e:
- log.warning("No providers are present or error when importing them! :%s", e)
- return
+ # Keeps list of providers keyed by module name and value is Tuple: version, provider_info
+ self._provider_directory: Dict[str, Tuple[str, Dict]] = OrderedDict()
+ self._provider_list: [Tuple[str, str, Dict]] = []
+ self._provider_set: Set[str] = set()
self._validator = _create_validator()
- self.__find_all_providers(providers.__path__)
+ # Local source folders are loaded first. They should take precedence over the package ones for
+ # 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.__find_all_airflow_builtin_providers_from_local_sources()
+ self.__find_all_providers_from_packages()
+ self.__creates_provider_directory()
+
+ def __creates_provider_directory(self):
+ """
+ Creates provider_directory as sorted (by package_name) OrderedDict.
+
+ Duplicates are removed from "package" providers in case corresponding "folder" provider is found.
+ The "folder" providers are from local sources (packages do not contain provider.yaml files),
+ so if someone has airflow installed from local sources, the providers are imported from there
+ first so, provider information should be taken from there.
+ :return:
+ """
+ self._provider_list.sort(key=lambda provider_list_element: provider_list_element[0])
+ for package_name, version, provider_info in self._provider_list:
+ self._provider_directory[package_name] = (version, provider_info)
+
+ def __find_all_providers_from_packages(self) -> None:
+ """
+ Finds 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.yaml.schema.json' schema.
- def __find_all_providers(self, paths: str):
- def onerror(_):
- exception_string = traceback.format_exc()
- log.warning(exception_string)
+ The providers are updated in the list of found providers.
- for module_info in pkgutil.walk_packages(paths, prefix="airflow.providers.", onerror=onerror):
+ """
+ for entry_point in pkg_resources.iter_entry_points('apache_airflow_provider'):
+ package_name = entry_point.dist.project_name
+ version = entry_point.dist.version
try:
- imported_module = importlib.import_module(module_info.name)
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when importing %s:%s", module_info.name, e)
+ provider_info = entry_point.load()()
+ except pkg_resources.VersionConflict as e:
+ log.warning(
+ "The provider package %s could not be registered because of version conflict : %s",
+ package_name,
+ e,
+ )
continue
- try:
- provider = importlib_resources.read_text(imported_module, 'provider.yaml')
- provider_info = yaml.safe_load(provider)
- self._validator.validate(provider_info)
- self._provider_directory[provider_info['package-name']] = provider_info
- except FileNotFoundError:
- # This is OK - this is not a provider package
- pass
- except TypeError as e:
- if "is not a package" not in str(e):
- log.warning("Error when loading 'provider.yaml' file from %s:%s}", module_info.name, e)
- # Otherwise this is OK - this is likely a module
- except Exception as e: # noqa pylint: disable=broad-except
- log.warning("Error when loading 'provider.yaml' file from %s:%s", module_info.name, e)
+ self._validator.validate(provider_info)
+ provider_info_package_name = provider_info['package-name']
+ if package_name != provider_info_package_name:
+ raise Exception(
+ f"The package '{package_name}' from setuptools and "
+ f"{provider_info_package_name} do not match. Please make sure they are"
+ f"aligned"
+ )
+ if package_name not in self._provider_set:
+ self._provider_set.add(package_name)
+ self._provider_list.append((package_name, version, provider_info))
+ else:
+ log.warning(
+ "The providers for package '%s' could not be registered because providers for that "
+ "package name have already been registered",
+ package_name,
+ )
+
+ def __find_all_airflow_builtin_providers_from_local_sources(self) -> None:
+ """
+ Finds all built-in airflow providers if airflow is run from the local sources.
+ It finds `provider.yaml` files for all such providers and registers the providers using those.
+
+ This 'provider.yaml' scanning takes precedence over scanning packages installed because
+ in case you have both sources and packages installed, the providers will be loaded from
+ the "airflow" sources rather than from the packages.
+ """
+ import airflow.providers
+
+ try:
+ for path in airflow.providers.__path__:
+ self.__add_provider_info_from_local_source_files_on_path(path)
+ except Exception as e: # noqa pylint: disable=broad-except
+ log.warning("Error when loading 'provider.yaml' files from airflow sources: %s", e)
+
+ def __add_provider_info_from_local_source_files_on_path(self, path) -> None:
+ """
+ Finds all the provider.yaml files in the directory specified.
+ :param path: path where to look for provider.yaml files
+ """
+ root_path = path
+ for folder, _, files in os.walk(path):
+ for filename in fnmatch.filter(files, "provider.yaml"):
+ self.__add_provider_info_from_local_source_file(filename, folder, root_path)
Review comment:
True. Especially on Mac
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org