Add discoverability for triggers in provider.yaml (#31576)

* Add discoverability for triggers in provider.yaml
apache · Jun 2, 2023 · dc5bf3f · dc5bf3f
1 parent a59076e
commit dc5bf3f
Show file tree

Hide file tree

Showing 11 changed files with 70 additions and 72 deletions.
diff --git a/airflow/provider.yaml.schema.json b/airflow/provider.yaml.schema.json
@@ -208,8 +208,8 @@
  "type": "string",
  "description": "Integration name. It must have a matching item in the 'integration' section of any provider."
  },
- "class-names": {
- "description": "List of the trigger class name that implements trigger.",
+ "python-modules": {
+ "description": "List of Python modules containing the triggers.",
  "type": "array",
  "items": {
  "type": "string"
@@ -219,7 +219,7 @@
  "additionalProperties": false,
  "required": [
  "integration-name",
- "class-names"
+ "python-modules"
  ]
  }
  },
@@ -312,7 +312,7 @@
  "items": {
  "type": "string"
  }
-  },
+ },
  "plugins": {
  "type": "array",
  "description": "Plugins exposed by the provider",

diff --git a/airflow/providers/amazon/provider.yaml b/airflow/providers/amazon/provider.yaml
@@ -505,6 +505,24 @@ hooks:
  python-modules:
  - airflow.providers.amazon.aws.hooks.appflow
 
+triggers:
+ - integration-name: AWS Batch
+ python-modules:
+ - airflow.providers.amazon.aws.triggers.batch
+ - integration-name: Amazon EC2
+ python-modules:
+ - airflow.providers.amazon.aws.triggers.ec2
+ - integration-name: Amazon Redshift
+ python-modules:
+ - airflow.providers.amazon.aws.triggers.redshift_cluster
+ - integration-name: Amazon SageMaker
+ python-modules:
+ - airflow.providers.amazon.aws.triggers.sagemaker
+ - integration-name: AWS Glue
+ python-modules:
+ - airflow.providers.amazon.aws.triggers.glue
+ - airflow.providers.amazon.aws.triggers.glue_crawler
+
 transfers:
  - source-integration-name: Amazon DynamoDB
  target-integration-name: Amazon Simple Storage Service (S3)

diff --git a/airflow/providers/apache/kafka/provider.yaml b/airflow/providers/apache/kafka/provider.yaml
@@ -58,8 +58,8 @@ sensors:
 
 triggers:
  - integration-name: Apache Kafka
- class-names:
- - airflow.providers.apache.kafka.triggers.await_message.AwaitMessageTrigger
+ python-modules:
+ - airflow.providers.apache.kafka.triggers.await_message
 
 connection-types:
  - hook-class-name: airflow.providers.apache.kafka.hooks.base.KafkaBaseHook

diff --git a/airflow/providers/apache/livy/provider.yaml b/airflow/providers/apache/livy/provider.yaml
@@ -70,8 +70,8 @@ hooks:
 
 triggers:
  - integration-name: Apache Livy
- class-names:
- - airflow.providers.apache.livy.triggers.livy.LivyTrigger
+ python-modules:
+ - airflow.providers.apache.livy.triggers.livy
 
 
 connection-types:

diff --git a/airflow/providers/cncf/kubernetes/provider.yaml b/airflow/providers/cncf/kubernetes/provider.yaml
@@ -109,8 +109,8 @@ hooks:
 
 triggers:
  - integration-name: Kubernetes
- class-names:
- - airflow.providers.cncf.kubernetes.triggers.pod.KubernetesPodTrigger
+ python-modules:
+ - airflow.providers.cncf.kubernetes.triggers.pod
 
 
 connection-types:

diff --git a/airflow/providers/databricks/provider.yaml b/airflow/providers/databricks/provider.yaml
@@ -98,8 +98,8 @@ hooks:
 
 triggers:
  - integration-name: Databricks
- class-names:
- - airflow.providers.databricks.triggers.databricks.DatabricksExecutionTrigger
+ python-modules:
+ - airflow.providers.databricks.triggers.databricks
 
 sensors:
  - integration-name: Databricks

diff --git a/airflow/providers/dbt/cloud/provider.yaml b/airflow/providers/dbt/cloud/provider.yaml
@@ -67,8 +67,8 @@ hooks:
 
 triggers:
  - integration-name: dbt Cloud
- class-names:
- - airflow.providers.dbt.cloud.triggers.dbt.DbtCloudRunJobTrigger
+ python-modules:
+ - airflow.providers.dbt.cloud.triggers.dbt
 
 
 connection-types:

diff --git a/airflow/providers/google/provider.yaml b/airflow/providers/google/provider.yaml
@@ -832,47 +832,35 @@ hooks:
 
 triggers:
  - integration-name: Google BigQuery Data Transfer Service
- class-names:
- - airflow.providers.google.cloud.triggers.bigquery_dts.BigQueryDataTransferRunTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.bigquery_dts
  - integration-name: Google BigQuery
- class-names:
- - airflow.providers.google.cloud.triggers.bigquery.BigQueryInsertJobTrigger
- - airflow.providers.google.cloud.triggers.bigquery.BigQueryCheckTrigger
- - airflow.providers.google.cloud.triggers.bigquery.BigQueryGetDataTrigger
- - airflow.providers.google.cloud.triggers.bigquery.BigQueryIntervalCheckTrigger
- - airflow.providers.google.cloud.triggers.bigquery.BigQueryValueCheckTrigger
- - airflow.providers.google.cloud.triggers.bigquery.BigQueryTableExistenceTrigger
- - airflow.providers.google.cloud.triggers.bigquery.BigQueryTablePartitionExistenceTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.bigquery
  - integration-name: Google Cloud Build
- class-names:
- - airflow.providers.google.cloud.triggers.cloud_build.CloudBuildCreateBuildTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.cloud_build
  - integration-name: Google Cloud Composer
- class-names:
- - airflow.providers.google.cloud.triggers.cloud_composer.CloudComposerExecutionTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.cloud_composer
  - integration-name: Google Dataflow
- class-names:
- - airflow.providers.google.cloud.triggers.dataflow.TemplateJobStartTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.dataflow
  - integration-name: Google Data Fusion
- class-names:
- - airflow.providers.google.cloud.triggers.datafusion.DataFusionStartPipelineTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.datafusion
  - integration-name: Google Dataproc
- class-names:
- - airflow.providers.google.cloud.triggers.dataproc.DataprocBaseTrigger
- - airflow.providers.google.cloud.triggers.dataproc.DataprocSubmitTrigger
- - airflow.providers.google.cloud.triggers.dataproc.DataprocClusterTrigger
- - airflow.providers.google.cloud.triggers.dataproc.DataprocBatchTrigger
- - airflow.providers.google.cloud.triggers.dataproc.DataprocDeleteClusterTrigger
- - airflow.providers.google.cloud.triggers.dataproc.DataprocWorkflowTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.dataproc
  - integration-name: Google Cloud Storage (GCS)
- class-names:
- - airflow.providers.google.cloud.triggers.gcs.GCSBlobTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.gcs
  - integration-name: Google Kubernetes Engine
- class-names:
- - airflow.providers.google.cloud.triggers.kubernetes_engine.GKEStartPodTrigger
- - airflow.providers.google.cloud.triggers.kubernetes_engine.GKEOperationTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.kubernetes_engine
  - integration-name: Google Machine Learning Engine
- class-names:
- - airflow.providers.google.cloud.triggers.mlengine.MLEngineStartTrainingJobTrigger
+ python-modules:
+ - airflow.providers.google.cloud.triggers.mlengine
 
 transfers:
  - source-integration-name: Presto

diff --git a/airflow/providers/microsoft/azure/provider.yaml b/airflow/providers/microsoft/azure/provider.yaml
@@ -218,6 +218,14 @@ hooks:
  python-modules:
  - airflow.providers.microsoft.azure.hooks.synapse
 
+triggers:
+ - integration-name: Microsoft Azure Data Factory
+ python-modules:
+ - airflow.providers.microsoft.azure.triggers.data_factory
+ - integration-name: Microsoft Azure Blob Storage
+ python-modules:
+ - airflow.providers.microsoft.azure.triggers.wasb
+
 transfers:
  - source-integration-name: Local
  target-integration-name: Microsoft Azure Data Lake Storage

diff --git a/airflow/providers_manager.py b/airflow/providers_manager.py
@@ -956,7 +956,7 @@ def initialize_providers_triggers(self):
  self.initialize_providers_list()
  for provider_package, provider in self._provider_dict.items():
  for trigger in provider.data.get("triggers", []):
- for trigger_class_name in trigger.get("class-names"):
+ for trigger_class_name in trigger.get("python-modules"):
  self._trigger_info_set.add(
  TriggerInfo(
  package_name=provider_package,

diff --git a/scripts/in_container/run_provider_yaml_files_check.py b/scripts/in_container/run_provider_yaml_files_check.py
@@ -44,6 +44,7 @@
 EXCLUDED_MODULES = [
  "airflow.providers.apache.hdfs.sensors.hdfs",
  "airflow.providers.apache.hdfs.hooks.hdfs",
+ "airflow.providers.cncf.kubernetes.triggers.kubernetes_pod",
 ]
 
 
@@ -229,10 +230,10 @@ def parse_module_data(provider_data, resource_type, yaml_file_path):
 
 
 def check_correctness_of_list_of_sensors_operators_hook_modules(yaml_files: dict[str, dict]):
- print("Checking completeness of list of {sensors, hooks, operators}")
- print(" -- {sensors, hooks, operators} - Expected modules (left) : Current modules (right)")
+ print("Checking completeness of list of {sensors, hooks, operators, triggers}")
+ print(" -- {sensors, hooks, operators, triggers} - Expected modules (left) : Current modules (right)")
  for (yaml_file_path, provider_data), resource_type in product(
- yaml_files.items(), ["sensors", "operators", "hooks"]
+ yaml_files.items(), ["sensors", "operators", "hooks", "triggers"]
  ):
  expected_modules, provider_package, resource_data = parse_module_data(
  provider_data, resource_type, yaml_file_path
@@ -254,9 +255,9 @@ def check_correctness_of_list_of_sensors_operators_hook_modules(yaml_files: dict
 
 
 def check_duplicates_in_integrations_names_of_hooks_sensors_operators(yaml_files: dict[str, dict]):
- print("Checking for duplicates in list of {sensors, hooks, operators}")
+ print("Checking for duplicates in list of {sensors, hooks, operators, triggers}")
  for (yaml_file_path, provider_data), resource_type in product(
- yaml_files.items(), ["sensors", "operators", "hooks"]
+ yaml_files.items(), ["sensors", "operators", "hooks", "triggers"]
  ):
  resource_data = provider_data.get(resource_type, [])
  current_integrations = [r.get("integration-name", "") for r in resource_data]
@@ -294,7 +295,7 @@ def check_completeness_of_list_of_transfers(yaml_files: dict[str, dict]):
  )
 
 
-def check_hook_classes(yaml_files: dict[str, dict]):
+def check_hook_connection_classes(yaml_files: dict[str, dict]):
  print("Checking connection classes belong to package, exist and are classes")
  resource_type = "hook-class-names"
  for yaml_file_path, provider_data in yaml_files.items():
@@ -306,22 +307,6 @@ def check_hook_classes(yaml_files: dict[str, dict]):
  )
 
 
-def check_trigger_classes(yaml_files: dict[str, dict]):
- print("Checking triggers classes belong to package, exist and are classes")
- resource_type = "triggers"
- for yaml_file_path, provider_data in yaml_files.items():
- provider_package = pathlib.Path(yaml_file_path).parent.as_posix().replace("/", ".")
- trigger_classes = {
- name
- for trigger_class in provider_data.get(resource_type, {})
- for name in trigger_class["class-names"]
- }
- if trigger_classes:
- check_if_objects_exist_and_belong_to_package(
- trigger_classes, provider_package, yaml_file_path, resource_type, ObjectType.CLASS
- )
-
-
 def check_plugin_classes(yaml_files: dict[str, dict]):
  print("Checking plugin classes belong to package, exist and are classes")
  resource_type = "plugins"
@@ -377,7 +362,7 @@ def check_invalid_integration(yaml_files: dict[str, dict]):
  all_integration_names = set(get_all_integration_names(yaml_files))
 
  for (yaml_file_path, provider_data), resource_type in product(
- yaml_files.items(), ["sensors", "operators", "hooks"]
+ yaml_files.items(), ["sensors", "operators", "hooks", "triggers"]
  ):
  resource_data = provider_data.get(resource_type, [])
  current_names = {r["integration-name"] for r in resource_data}
@@ -522,10 +507,9 @@ def check_providers_have_all_documentation_files(yaml_files: dict[str, dict]):
 
  check_completeness_of_list_of_transfers(all_parsed_yaml_files)
  check_duplicates_in_list_of_transfers(all_parsed_yaml_files)
- check_hook_classes(all_parsed_yaml_files)
+ check_hook_connection_classes(all_parsed_yaml_files)
  check_plugin_classes(all_parsed_yaml_files)
  check_extra_link_classes(all_parsed_yaml_files)
- check_trigger_classes(all_parsed_yaml_files)
  check_correctness_of_list_of_sensors_operators_hook_modules(all_parsed_yaml_files)
  check_unique_provider_name(all_parsed_yaml_files)
  check_providers_have_all_documentation_files(all_parsed_yaml_files)