How to make Airflow Sensors succeed given it meets a timeout?

user_7841401 0
Viewed 436

Referring to this, Airflow sensors allow us to check a criteria before running the next tasks. Is there a way to mark successfully terminate the sensor given a user puts a timeout and another flag for it?

In my use case, I am having to check a condition via a sensor but only during a particular time frame post which I would want the DAG / followings tasks to run normally.

airflow-2.x airflow-scheduler airflow
1 Answers

You can do that by creating a custom sensor class. You will need to override the poke function and place the logic you wish to set. For example:

from airflow.sensors.sql import SqlSensor

class MySqlSensor(SqlSensor):

    def is_time_frame(self):
        # TODO: implement a function that returns True if we want to ignore the sensor

    def poke(self, context):
        if self.is_time_frame():
            return True
        super().poke(context)

In this example when sensor is poking it first check the time window. If current time is within the window then the sensor will return True and exit. for any other case the sensor will do it's work - In that specific example running a SQL query until the query returns True.

user_146244090
Related