| #!/usr/bin/env python |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| from __future__ import annotations |
| |
| import functools |
| import importlib |
| import inspect |
| import itertools |
| import json |
| import os |
| import pathlib |
| import platform |
| import sys |
| import textwrap |
| import warnings |
| from collections import Counter |
| from enum import Enum |
| from typing import Any, Callable, Iterable |
| |
| import jsonschema |
| import yaml |
| from jsonpath_ng.ext import parse |
| from rich.console import Console |
| from tabulate import tabulate |
| |
| from airflow.cli.commands.info_command import Architecture |
| from airflow.exceptions import AirflowProviderDeprecationWarning |
| from airflow.providers_manager import ProvidersManager |
| |
| # Those are deprecated modules that contain removed Hooks/Sensors/Operators that we left in the code |
| # so that users can get a very specific error message when they try to use them. |
| |
| DEPRECATED_MODULES = [ |
| "airflow.providers.apache.hdfs.sensors.hdfs", |
| "airflow.providers.apache.hdfs.hooks.hdfs", |
| "airflow.providers.cncf.kubernetes.triggers.kubernetes_pod", |
| "airflow.providers.cncf.kubernetes.operators.kubernetes_pod", |
| "airflow.providers.tabular.hooks.tabular", |
| ] |
| |
| KNOWN_DEPRECATED_CLASSES = [ |
| "airflow.providers.google.cloud.links.dataproc.DataprocLink", |
| ] |
| |
| try: |
| from yaml import CSafeLoader as SafeLoader |
| except ImportError: |
| from yaml import SafeLoader # type: ignore |
| |
| if __name__ != "__main__": |
| raise SystemExit( |
| "This file is intended to be executed as an executable program. You cannot use it as a module." |
| ) |
| |
| ROOT_DIR = pathlib.Path(__file__).resolve().parents[2] |
| DOCS_DIR = ROOT_DIR.joinpath("docs") |
| PROVIDER_DATA_SCHEMA_PATH = ROOT_DIR.joinpath("airflow", "provider.yaml.schema.json") |
| PROVIDER_ISSUE_TEMPLATE_PATH = ROOT_DIR.joinpath( |
| ".github", "ISSUE_TEMPLATE", "airflow_providers_bug_report.yml" |
| ) |
| CORE_INTEGRATIONS = ["SQL", "Local"] |
| |
| errors: list[str] = [] |
| |
| console = Console(width=400, color_system="standard") |
| # you need to enable warnings for all deprecations - needed by importlib library to show deprecations |
| if os.environ.get("PYTHONWARNINGS") != "default": |
| console.print( |
| "[red]Error: PYTHONWARNINGS not set[/]\n" |
| "You must set `PYTHONWARNINGS=default` environment variable to run this script" |
| ) |
| sys.exit(1) |
| |
| suspended_providers: set[str] = set() |
| suspended_logos: set[str] = set() |
| suspended_integrations: set[str] = set() |
| |
| |
| def _filepath_to_module(filepath: pathlib.Path) -> str: |
| p = filepath.resolve().relative_to(ROOT_DIR).as_posix() |
| if p.endswith(".py"): |
| p = p[:-3] |
| return p.replace("/", ".") |
| |
| |
| def _load_schema() -> dict[str, Any]: |
| with PROVIDER_DATA_SCHEMA_PATH.open() as schema_file: |
| content = json.load(schema_file) |
| return content |
| |
| |
| def _load_package_data(package_paths: Iterable[str]): |
| schema = _load_schema() |
| result = {} |
| for provider_yaml_path in package_paths: |
| with open(provider_yaml_path) as yaml_file: |
| provider = yaml.load(yaml_file, SafeLoader) |
| rel_path = pathlib.Path(provider_yaml_path).relative_to(ROOT_DIR).as_posix() |
| try: |
| jsonschema.validate(provider, schema=schema) |
| except jsonschema.ValidationError as ex: |
| msg = f"Unable to parse: {provider_yaml_path}. Original error {type(ex).__name__}: {ex}" |
| raise RuntimeError(msg) |
| if not provider["state"] == "suspended": |
| result[rel_path] = provider |
| else: |
| suspended_providers.add(provider["package-name"]) |
| for integration in provider["integrations"]: |
| suspended_integrations.add(integration["integration-name"]) |
| if "logo" in integration: |
| suspended_logos.add(integration["logo"]) |
| return result |
| |
| |
| def get_all_integration_names(yaml_files) -> list[str]: |
| all_integrations = [ |
| i["integration-name"] for f in yaml_files.values() if "integrations" in f for i in f["integrations"] |
| ] |
| all_integrations += ["Local"] |
| return all_integrations |
| |
| |
| def assert_sets_equal( |
| set1: set[str], |
| set_name_1: str, |
| set2: set[str], |
| set_name_2: str, |
| allow_extra_in_set2=False, |
| extra_message: str = "", |
| ): |
| try: |
| difference1 = set1.difference(set2) |
| except TypeError as e: |
| raise AssertionError(f"invalid type when attempting set difference: {e}") |
| except AttributeError as e: |
| raise AssertionError(f"first argument does not support set difference: {e}") |
| |
| try: |
| difference2 = set2.difference(set1) |
| except TypeError as e: |
| raise AssertionError(f"invalid type when attempting set difference: {e}") |
| except AttributeError as e: |
| raise AssertionError(f"second argument does not support set difference: {e}") |
| |
| if difference1 or (difference2 and not allow_extra_in_set2): |
| lines = [] |
| lines.append(f" Left set:{set_name_1}") |
| lines.append(f" Right set:{set_name_2}") |
| if difference1: |
| lines.append(" Items in the left set but not the right:") |
| for item in sorted(difference1): |
| lines.append(f" {item!r}") |
| if difference2 and not allow_extra_in_set2: |
| lines.append(" Items in the right set but not the left:") |
| for item in sorted(difference2): |
| lines.append(f" {item!r}") |
| |
| standard_msg = "\n".join(lines) |
| if extra_message: |
| standard_msg += f"\n{extra_message}" |
| raise AssertionError(standard_msg) |
| |
| |
| class ObjectType(Enum): |
| MODULE = "module" |
| CLASS = "class" |
| |
| |
| def check_if_object_exist( |
| object_name: str, resource_type: str, yaml_file_path: str, object_type: ObjectType |
| ) -> int: |
| num_errors = 0 |
| try: |
| if object_type == ObjectType.CLASS: |
| module_name, class_name = object_name.rsplit(".", maxsplit=1) |
| with warnings.catch_warnings(record=True) as w: |
| the_class = getattr(importlib.import_module(module_name), class_name) |
| for warn in w: |
| if warn.category == AirflowProviderDeprecationWarning: |
| if object_name in KNOWN_DEPRECATED_CLASSES: |
| console.print( |
| f"[yellow]The {object_name} class is deprecated and we know about it. " |
| f"It should be removed in the future." |
| ) |
| continue |
| errors.append( |
| f"The `{class_name}` class in {resource_type} list in {yaml_file_path} " |
| f"is deprecated with this message: '{warn.message}'.\n" |
| f"[yellow]How to fix it[/]: Please remove it from provider.yaml and replace with " |
| f"the new class." |
| ) |
| num_errors += 1 |
| if the_class and inspect.isclass(the_class): |
| return num_errors |
| elif object_type == ObjectType.MODULE: |
| with warnings.catch_warnings(record=True) as w: |
| module = importlib.import_module(object_name) |
| for warn in w: |
| if warn.category == AirflowProviderDeprecationWarning: |
| errors.append( |
| f"The `{object_name}` module in {resource_type} list in {yaml_file_path} " |
| f"is deprecated with this message: '{warn.message}'.\n" |
| f"[yellow]How to fix it[/]: Please remove it from provider.yaml and replace it " |
| f"with the new module. If you see warnings in classes - fix the classes so that " |
| f"they are not raising Deprecation Warnings when module is imported." |
| ) |
| num_errors += 1 |
| if inspect.ismodule(module): |
| return num_errors |
| else: |
| raise RuntimeError(f"Wrong enum {object_type}???") |
| except Exception as e: |
| errors.append( |
| f"The `{object_name}` object in {resource_type} list in {yaml_file_path} does not exist " |
| f"or is not a {object_type.value}: {e}" |
| ) |
| num_errors += 1 |
| else: |
| errors.append( |
| f"The `{object_name}` object in {resource_type} list in {yaml_file_path} does not exist " |
| f"or is not a {object_type.value}." |
| ) |
| num_errors += 1 |
| return num_errors |
| |
| |
| def check_if_objects_exist_and_belong_to_package( |
| object_names: set[str], |
| provider_package: str, |
| yaml_file_path: str, |
| resource_type: str, |
| object_type: ObjectType, |
| ) -> int: |
| num_errors = 0 |
| for object_name in object_names: |
| if os.environ.get("VERBOSE"): |
| console.print( |
| f"[bright_blue]Checking if {object_name} of {resource_type} " |
| f"in {yaml_file_path} is {object_type.value} and belongs to {provider_package} package" |
| ) |
| if not object_name.startswith(provider_package): |
| errors.append( |
| f"The `{object_name}` object in {resource_type} list in {yaml_file_path} does not start" |
| f" with the expected {provider_package}." |
| ) |
| num_errors += 1 |
| num_errors += check_if_object_exist(object_name, resource_type, yaml_file_path, object_type) |
| return num_errors |
| |
| |
| def parse_module_data(provider_data, resource_type, yaml_file_path): |
| package_dir = ROOT_DIR.joinpath(yaml_file_path).parent |
| provider_package = pathlib.Path(yaml_file_path).parent.as_posix().replace("/", ".") |
| py_files = itertools.chain( |
| package_dir.glob(f"**/{resource_type}/*.py"), |
| package_dir.glob(f"{resource_type}/*.py"), |
| package_dir.glob(f"**/{resource_type}/**/*.py"), |
| package_dir.glob(f"{resource_type}/**/*.py"), |
| ) |
| expected_modules = {_filepath_to_module(f) for f in py_files if f.name != "__init__.py"} |
| resource_data = provider_data.get(resource_type, []) |
| return expected_modules, provider_package, resource_data |
| |
| |
| def run_check(title: str): |
| def inner(func: Callable[..., tuple[int, int]]): |
| @functools.wraps(func) |
| def wrapper(*args, **kwargs): |
| console.print(f"\n[magenta]Starting check:[/] {title}\n") |
| check_num, error_num = func(*args, **kwargs) |
| console.print( |
| f"\n[magenta]Finished check:[/] Found {error_num} " |
| f"error{'' if error_num == 1 else 's'} in {check_num} " |
| f"checked items\n" |
| ) |
| return check_num, error_num |
| |
| return wrapper |
| |
| return inner |
| |
| |
| @run_check("Checking integration duplicates") |
| def check_integration_duplicates(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| """Integration names must be globally unique.""" |
| num_errors = 0 |
| all_integrations = get_all_integration_names(yaml_files) |
| num_integrations = len(all_integrations) |
| duplicates = [(k, v) for (k, v) in Counter(all_integrations).items() if v > 1] |
| |
| if duplicates: |
| console.print( |
| "Duplicate integration names found. Integration names must be globally unique. " |
| "Please delete duplicates." |
| ) |
| errors.append(tabulate(duplicates, headers=["Integration name", "Number of occurrences"])) |
| num_errors += 1 |
| return num_integrations, num_errors |
| |
| |
| @run_check("Checking completeness of list of {sensors, hooks, operators, triggers}") |
| def check_correctness_of_list_of_sensors_operators_hook_trigger_modules( |
| yaml_files: dict[str, dict], |
| ) -> tuple[int, int]: |
| num_errors = 0 |
| num_modules = 0 |
| for (yaml_file_path, provider_data), resource_type in itertools.product( |
| yaml_files.items(), ["sensors", "operators", "hooks", "triggers"] |
| ): |
| expected_modules, provider_package, resource_data = parse_module_data( |
| provider_data, resource_type, yaml_file_path |
| ) |
| expected_modules = {module for module in expected_modules if module not in DEPRECATED_MODULES} |
| current_modules = {str(i) for r in resource_data for i in r.get("python-modules", [])} |
| num_modules += len(current_modules) |
| num_errors += check_if_objects_exist_and_belong_to_package( |
| current_modules, provider_package, yaml_file_path, resource_type, ObjectType.MODULE |
| ) |
| try: |
| package_name = os.fspath(ROOT_DIR.joinpath(yaml_file_path).parent.relative_to(ROOT_DIR)).replace( |
| "/", "." |
| ) |
| assert_sets_equal( |
| set(expected_modules), |
| f"Found list of {resource_type} modules in provider package: {package_name}", |
| set(current_modules), |
| f"Currently configured list of {resource_type} modules in {yaml_file_path}", |
| extra_message="[yellow]Additional check[/]: If there are deprecated modules in the list," |
| "please add them to DEPRECATED_MODULES in " |
| f"{pathlib.Path(__file__).relative_to(ROOT_DIR)}", |
| ) |
| except AssertionError as ex: |
| nested_error = textwrap.indent(str(ex), " ") |
| errors.append( |
| f"Incorrect content of key '{resource_type}/python-modules' " |
| f"in file: {yaml_file_path}\n{nested_error}" |
| ) |
| num_errors += 1 |
| return num_modules, num_errors |
| |
| |
| @run_check("Checking for duplicates in list of {sensors, hooks, operators, triggers}") |
| def check_duplicates_in_integrations_names_of_hooks_sensors_operators( |
| yaml_files: dict[str, dict], |
| ) -> tuple[int, int]: |
| num_errors = 0 |
| num_integrations = 0 |
| for (yaml_file_path, provider_data), resource_type in itertools.product( |
| yaml_files.items(), ["sensors", "operators", "hooks", "triggers"] |
| ): |
| resource_data = provider_data.get(resource_type, []) |
| count_integrations = Counter(r.get("integration-name", "") for r in resource_data) |
| num_integrations += len(count_integrations) |
| for integration, count in count_integrations.items(): |
| if count > 1: |
| errors.append( |
| f"Duplicated content of '{resource_type}/integration-name/{integration}' " |
| f"in file: {yaml_file_path}" |
| ) |
| num_errors += 1 |
| return num_integrations, num_errors |
| |
| |
| @run_check("Checking completeness of list of transfers") |
| def check_completeness_of_list_of_transfers(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| resource_type = "transfers" |
| num_errors = 0 |
| num_transfers = 0 |
| for yaml_file_path, provider_data in yaml_files.items(): |
| expected_modules, provider_package, resource_data = parse_module_data( |
| provider_data, resource_type, yaml_file_path |
| ) |
| expected_modules = {module for module in expected_modules if module not in DEPRECATED_MODULES} |
| current_modules = {r.get("python-module") for r in resource_data} |
| num_transfers += len(current_modules) |
| num_errors += check_if_objects_exist_and_belong_to_package( |
| current_modules, provider_package, yaml_file_path, resource_type, ObjectType.MODULE |
| ) |
| try: |
| package_name = os.fspath(ROOT_DIR.joinpath(yaml_file_path).parent.relative_to(ROOT_DIR)).replace( |
| "/", "." |
| ) |
| assert_sets_equal( |
| set(expected_modules), |
| f"Found list of transfer modules in provider package: {package_name}", |
| set(current_modules), |
| f"Currently configured list of transfer modules in {yaml_file_path}", |
| ) |
| except AssertionError as ex: |
| nested_error = textwrap.indent(str(ex), " ") |
| errors.append( |
| f"Incorrect content of key '{resource_type}/python-module' " |
| f"in file: {yaml_file_path}\n{nested_error}" |
| ) |
| num_errors += 1 |
| return num_transfers, num_errors |
| |
| |
| @run_check("Checking if hook classes specified by hook-class-name in connection type are importable") |
| def check_hook_class_name_entries_in_connection_types(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| resource_type = "connection-types" |
| num_errors = 0 |
| num_connection_types = 0 |
| for yaml_file_path, provider_data in yaml_files.items(): |
| provider_package = pathlib.Path(yaml_file_path).parent.as_posix().replace("/", ".") |
| connection_types = provider_data.get(resource_type) |
| if connection_types: |
| num_connection_types += len(connection_types) |
| hook_class_names = {connection_type["hook-class-name"] for connection_type in connection_types} |
| num_errors += check_if_objects_exist_and_belong_to_package( |
| hook_class_names, provider_package, yaml_file_path, resource_type, ObjectType.CLASS |
| ) |
| return num_connection_types, num_errors |
| |
| |
| @run_check("Checking plugin classes belong to package are importable and belong to package") |
| def check_plugin_classes(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| resource_type = "plugins" |
| num_errors = 0 |
| num_plugins = 0 |
| for yaml_file_path, provider_data in yaml_files.items(): |
| provider_package = pathlib.Path(yaml_file_path).parent.as_posix().replace("/", ".") |
| plugins = provider_data.get(resource_type) |
| if plugins: |
| num_plugins += len(plugins) |
| num_errors += check_if_objects_exist_and_belong_to_package( |
| {plugin["plugin-class"] for plugin in plugins}, |
| provider_package, |
| yaml_file_path, |
| resource_type, |
| ObjectType.CLASS, |
| ) |
| return num_plugins, num_errors |
| |
| |
| @run_check("Checking extra-links belong to package, exist and are classes") |
| def check_extra_link_classes(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| resource_type = "extra-links" |
| num_errors = 0 |
| num_extra_links = 0 |
| for yaml_file_path, provider_data in yaml_files.items(): |
| provider_package = pathlib.Path(yaml_file_path).parent.as_posix().replace("/", ".") |
| extra_links = provider_data.get(resource_type) |
| if extra_links: |
| num_extra_links += len(extra_links) |
| num_errors += check_if_objects_exist_and_belong_to_package( |
| extra_links, provider_package, yaml_file_path, resource_type, ObjectType.CLASS |
| ) |
| return num_extra_links, num_errors |
| |
| |
| @run_check("Checking notifications belong to package, exist and are classes") |
| def check_notification_classes(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| resource_type = "notifications" |
| num_errors = 0 |
| num_notifications = 0 |
| for yaml_file_path, provider_data in yaml_files.items(): |
| provider_package = pathlib.Path(yaml_file_path).parent.as_posix().replace("/", ".") |
| notifications = provider_data.get(resource_type) |
| if notifications: |
| num_notifications += len(notifications) |
| num_errors += check_if_objects_exist_and_belong_to_package( |
| notifications, provider_package, yaml_file_path, resource_type, ObjectType.CLASS |
| ) |
| return num_notifications, num_errors |
| |
| |
| @run_check("Checking for duplicates in list of transfers") |
| def check_duplicates_in_list_of_transfers(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| resource_type = "transfers" |
| num_errors = 0 |
| num_integrations = 0 |
| for yaml_file_path, provider_data in yaml_files.items(): |
| resource_data = provider_data.get(resource_type, []) |
| count_integrations = Counter( |
| (r.get("source-integration-name", ""), r.get("target-integration-name", "")) |
| for r in resource_data |
| ) |
| num_integrations += len(count_integrations) |
| for (source, target), count in count_integrations.items(): |
| if count > 1: |
| errors.append( |
| f"Duplicated content of \n" |
| f" '{resource_type}/source-integration-name/{source}' " |
| f" '{resource_type}/target-integration-name/{target}' " |
| f"in file: {yaml_file_path}" |
| ) |
| num_errors += 1 |
| return num_integrations, num_errors |
| |
| |
| @run_check("Detect unregistered integrations") |
| def check_invalid_integration(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| all_integration_names = set(get_all_integration_names(yaml_files)) |
| num_errors = 0 |
| num_integrations = len(all_integration_names) |
| for (yaml_file_path, provider_data), resource_type in itertools.product( |
| yaml_files.items(), ["sensors", "operators", "hooks", "triggers"] |
| ): |
| resource_data = provider_data.get(resource_type, []) |
| current_names = {r["integration-name"] for r in resource_data} |
| invalid_names = current_names - all_integration_names |
| if invalid_names: |
| errors.append( |
| f"Incorrect content of key '{resource_type}/integration-name' in file: {yaml_file_path}. " |
| f"Invalid values: {invalid_names}" |
| ) |
| num_errors += 1 |
| |
| for (yaml_file_path, provider_data), key in itertools.product( |
| yaml_files.items(), ["source-integration-name", "target-integration-name"] |
| ): |
| resource_data = provider_data.get("transfers", []) |
| current_names = {r[key] for r in resource_data} |
| invalid_names = current_names - all_integration_names - suspended_integrations |
| if invalid_names: |
| errors.append( |
| f"Incorrect content of key 'transfers/{key}' in file: {yaml_file_path}. " |
| f"Invalid values: {invalid_names}" |
| ) |
| num_errors += 1 |
| return num_integrations, num_errors |
| |
| |
| @run_check("Checking doc files") |
| def check_doc_files(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| num_docs = 0 |
| num_errors = 0 |
| current_doc_urls: list[str] = [] |
| current_logo_urls: list[str] = [] |
| for provider in yaml_files.values(): |
| if "integrations" in provider: |
| current_doc_urls.extend( |
| guide |
| for guides in provider["integrations"] |
| if "how-to-guide" in guides |
| for guide in guides["how-to-guide"] |
| ) |
| current_logo_urls.extend( |
| integration["logo"] for integration in provider["integrations"] if "logo" in integration |
| ) |
| if "transfers" in provider: |
| current_doc_urls.extend( |
| op["how-to-guide"] for op in provider["transfers"] if "how-to-guide" in op |
| ) |
| if suspended_providers: |
| console.print("[yellow]Suspended/Removed providers:[/]") |
| console.print(suspended_providers) |
| |
| expected_doc_files = itertools.chain( |
| DOCS_DIR.glob("apache-airflow-providers-*/operators/**/*.rst"), |
| DOCS_DIR.glob("apache-airflow-providers-*/transfer/**/*.rst"), |
| ) |
| |
| expected_doc_urls = { |
| f"/docs/{f.relative_to(DOCS_DIR).as_posix()}" |
| for f in expected_doc_files |
| if f.name != "index.rst" |
| and "_partials" not in f.parts |
| and not f.relative_to(DOCS_DIR).as_posix().startswith(tuple(suspended_providers)) |
| } | { |
| f"/docs/{f.relative_to(DOCS_DIR).as_posix()}" |
| for f in DOCS_DIR.glob("apache-airflow-providers-*/operators.rst") |
| if not f.relative_to(DOCS_DIR).as_posix().startswith(tuple(suspended_providers)) |
| } |
| if suspended_logos: |
| console.print("[yellow]Suspended logos:[/]") |
| console.print(suspended_logos) |
| console.print() |
| expected_logo_urls = { |
| f"/{f.relative_to(DOCS_DIR).as_posix()}" |
| for f in (DOCS_DIR / "integration-logos").rglob("*") |
| if f.is_file() and not f"/{f.relative_to(DOCS_DIR).as_posix()}".startswith(tuple(suspended_logos)) |
| } |
| |
| try: |
| console.print("Checking document urls") |
| assert_sets_equal( |
| set(expected_doc_urls), |
| "Document urls found in airflow/docs", |
| set(current_doc_urls), |
| "Document urls configured in provider.yaml files", |
| ) |
| console.print(f"Checked {len(current_doc_urls)} doc urls") |
| console.print() |
| console.print("Checking logo urls") |
| assert_sets_equal( |
| set(expected_logo_urls), |
| "Logo urls found in airflow/docs/integration-logos", |
| set(current_logo_urls), |
| "Logo urls configured in provider.yaml files", |
| ) |
| console.print(f"Checked {len(current_logo_urls)} logo urls") |
| console.print() |
| except AssertionError as ex: |
| nested_error = textwrap.indent(str(ex), " ") |
| errors.append( |
| f"Discrepancies between documentation/logos for providers and provider.yaml files " |
| f"[yellow]How to fix it[/]: Please synchronize the docs/logs.\n{nested_error}" |
| ) |
| num_errors += 1 |
| return num_docs, num_errors |
| |
| |
| @run_check("Checking if provider names are unique") |
| def check_unique_provider_name(yaml_files: dict[str, dict]) -> tuple[int, int]: |
| num_errors = 0 |
| name_counter = Counter(d["name"] for d in yaml_files.values()) |
| duplicates = {k for k, v in name_counter.items() if v > 1} |
| if duplicates: |
| errors.append(f"Provider name must be unique. Duplicates: {duplicates}") |
| num_errors += 1 |
| return len(name_counter.items()), num_errors |
| |
| |
| @run_check(f"Checking providers are mentioned in {PROVIDER_ISSUE_TEMPLATE_PATH}") |
| def check_providers_are_mentioned_in_issue_template(yaml_files: dict[str, dict]): |
| num_errors = 0 |
| num_providers = 0 |
| prefix_len = len("apache-airflow-providers-") |
| short_provider_names = [d["package-name"][prefix_len:] for d in yaml_files.values()] |
| # exclude deprecated provider that shouldn't be in issue template |
| deprecated_providers: list[str] = [] |
| for item in deprecated_providers: |
| short_provider_names.remove(item) |
| num_providers += len(short_provider_names) |
| jsonpath_expr = parse('$.body[?(@.attributes.label == "Apache Airflow Provider(s)")]..options[*]') |
| with PROVIDER_ISSUE_TEMPLATE_PATH.open() as issue_file: |
| issue_template = yaml.safe_load(issue_file) |
| all_mentioned_providers = [match.value for match in jsonpath_expr.find(issue_template)] |
| try: |
| # in case of suspended providers, we still want to have them in the issue template |
| assert_sets_equal( |
| set(short_provider_names), |
| "Provider names found in provider.yaml files", |
| set(all_mentioned_providers), |
| f"Provider names mentioned in {PROVIDER_ISSUE_TEMPLATE_PATH}", |
| allow_extra_in_set2=True, |
| ) |
| except AssertionError as ex: |
| nested_error = textwrap.indent(str(ex), " ") |
| errors.append( |
| f"Discrepancies between providers available in `airflow/providers` and providers " |
| f"in {PROVIDER_ISSUE_TEMPLATE_PATH}.\n" |
| f"[yellow]How to fix it[/]: Please synchronize the list.\n{nested_error}" |
| ) |
| num_errors += 1 |
| return num_providers, num_errors |
| |
| |
| @run_check("Checking providers have all documentation files") |
| def check_providers_have_all_documentation_files(yaml_files: dict[str, dict]): |
| num_errors = 0 |
| num_providers = 0 |
| expected_files = ["commits.rst", "index.rst", "installing-providers-from-sources.rst"] |
| for package_info in yaml_files.values(): |
| num_providers += 1 |
| package_name = package_info["package-name"] |
| provider_dir = DOCS_DIR.joinpath(package_name) |
| for file in expected_files: |
| if not provider_dir.joinpath(file).is_file(): |
| errors.append( |
| f"The provider {package_name} misses `{file}` in documentation. " |
| f"[yellow]How to fix it[/]: Please add the file to {provider_dir}" |
| ) |
| num_errors += 1 |
| return num_providers, num_errors |
| |
| |
| if __name__ == "__main__": |
| ProvidersManager().initialize_providers_configuration() |
| architecture = Architecture.get_current() |
| console.print(f"Verifying packages on {architecture} architecture. Platform: {platform.machine()}.") |
| provider_files_pattern = pathlib.Path(ROOT_DIR, "airflow", "providers").rglob("provider.yaml") |
| all_provider_files = sorted(str(path) for path in provider_files_pattern) |
| if len(sys.argv) > 1: |
| paths = [os.fspath(ROOT_DIR / f) for f in sorted(sys.argv[1:])] |
| else: |
| paths = all_provider_files |
| |
| all_parsed_yaml_files: dict[str, dict] = _load_package_data(paths) |
| |
| all_files_loaded = len(all_provider_files) == len(paths) |
| check_integration_duplicates(all_parsed_yaml_files) |
| check_duplicates_in_list_of_transfers(all_parsed_yaml_files) |
| check_duplicates_in_integrations_names_of_hooks_sensors_operators(all_parsed_yaml_files) |
| |
| check_completeness_of_list_of_transfers(all_parsed_yaml_files) |
| check_hook_class_name_entries_in_connection_types(all_parsed_yaml_files) |
| check_plugin_classes(all_parsed_yaml_files) |
| check_extra_link_classes(all_parsed_yaml_files) |
| check_correctness_of_list_of_sensors_operators_hook_trigger_modules(all_parsed_yaml_files) |
| check_notification_classes(all_parsed_yaml_files) |
| check_unique_provider_name(all_parsed_yaml_files) |
| check_providers_have_all_documentation_files(all_parsed_yaml_files) |
| |
| if all_files_loaded: |
| # Only check those if all provider files are loaded |
| check_doc_files(all_parsed_yaml_files) |
| check_invalid_integration(all_parsed_yaml_files) |
| check_providers_are_mentioned_in_issue_template(all_parsed_yaml_files) |
| |
| if errors: |
| error_num = len(errors) |
| console.print(f"[red]Found {error_num} error{'' if error_num == 1 else 's'} in providers[/]") |
| for error in errors: |
| console.print(f"[red]Error:[/] {error}") |
| sys.exit(1) |
