phanikumv commented on code in PR #30074:
URL: https://github.com/apache/airflow/pull/30074#discussion_r1138274773
##########
airflow/providers/microsoft/azure/triggers/data_factory.py:
##########
@@ -89,3 +89,119 @@ async def run(self) -> AsyncIterator["TriggerEvent"]:
await asyncio.sleep(self.poke_interval)
except Exception as e:
yield TriggerEvent({"status": "error", "message": str(e)})
+
+
+class AzureDataFactoryTrigger(BaseTrigger):
+ """
+ AzureDataFactoryTrigger is triggered when Azure data factory pipeline job
succeeded or failed.
+ When wait_for_termination is set to False it triggered immediately with
success status
+
+ :param run_id: Run id of a Azure data pipeline run job.
+ :param azure_data_factory_conn_id: The connection identifier for
connecting to Azure Data Factory.
+ :param end_time: Time in seconds when triggers will timeout.
+ :param resource_group_name: The resource group name.
+ :param factory_name: The data factory name.
+ :param wait_for_termination: Flag to wait on a pipeline run's termination.
+ :param check_interval: Time in seconds to check on a pipeline run's status.
+ """
+
+ QUEUED = "Queued"
+ IN_PROGRESS = "InProgress"
+ SUCCEEDED = "Succeeded"
+ FAILED = "Failed"
+ CANCELING = "Canceling"
+ CANCELLED = "Cancelled"
+
+ INTERMEDIATE_STATES: list[str] = [QUEUED, IN_PROGRESS, CANCELING]
+ FAILURE_STATES: list[str] = [FAILED, CANCELLED]
+ SUCCESS_STATES: list[str] = [SUCCEEDED]
+ TERMINAL_STATUSES: list[str] = [CANCELLED, FAILED, SUCCEEDED]
+
+ def __init__(
+ self,
+ run_id: str,
+ azure_data_factory_conn_id: str,
+ end_time: float,
+ resource_group_name: str | None = None,
+ factory_name: str | None = None,
+ wait_for_termination: bool = True,
+ check_interval: int = 60,
+ ):
+ super().__init__()
+ self.azure_data_factory_conn_id = azure_data_factory_conn_id
+ self.check_interval = check_interval
+ self.run_id = run_id
+ self.wait_for_termination = wait_for_termination
+ self.resource_group_name = resource_group_name
+ self.factory_name = factory_name
+ self.end_time = end_time
+
+ def serialize(self) -> tuple[str, dict[str, Any]]:
+ """Serializes AzureDataFactoryTrigger arguments and classpath."""
+ return (
+
"airflow.providers.microsoft.azure.triggers.data_factory.AzureDataFactoryTrigger",
+ {
+ "azure_data_factory_conn_id": self.azure_data_factory_conn_id,
+ "check_interval": self.check_interval,
+ "run_id": self.run_id,
+ "wait_for_termination": self.wait_for_termination,
+ "resource_group_name": self.resource_group_name,
+ "factory_name": self.factory_name,
+ "end_time": self.end_time,
+ },
+ )
+
+ async def run(self) -> AsyncIterator["TriggerEvent"]:
+ """Make async connection to Azure Data Factory, polls for the pipeline
run status"""
+ hook =
AzureDataFactoryAsyncHook(azure_data_factory_conn_id=self.azure_data_factory_conn_id)
+ try:
+ pipeline_status = await hook.get_adf_pipeline_run_status(
+ run_id=self.run_id,
+ resource_group_name=self.resource_group_name,
+ factory_name=self.factory_name,
+ )
+ if self.wait_for_termination:
Review Comment:
No
When wait_for_termination=True and deferrable=False, we submit and "poll" on
the worker
When wait_for_termination=True and deferrable=True, we submit and "defer"
using Triggerer
When wait_for_termination=False and deferrable=False, we only submit
When wait_for_termination=False and deferrable=True, we only submit and no
deferrable takes place <--- this is where we should warn saying
"Deferrable=True" does not have any effect as wait_for_termination=False
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
