I have the following custom Airflow Operator (Class) that is pretty much checking if a query returns True, if it does then it runs that task successfully and continue running the next task. If False, then it returns a SkipException. I also have a soft_fail set to True which also skips if False but in case someone forgets to add that in when using this operator I wanted to add it directly in the code.
Pythonically, is this an acceptable way of writing this if/else statement?
class SnowDataSensor(SqlSensor):
def poke(self, context):
hook = SnowflakeHook(warehouse='warehouse')
self.log.info('Poking: %s', self.sql)
record = hook.execute_query(self.sql)
logging.info(f'record: {record}')
if not record[0]:
raise AirflowSkipException
else:
return True
This operator will be called like this as a task:
tables_sensor = SnowDataSensor(
conn_id='snow_default',
soft_fail=True,
task_id=f'data_exists',
mode='reschedule',
poke_interval=60, # 60 sec
retries=1,
retry_delay=timedelta(minutes=10),
sql=f"""
SELECT COUNT(*) > 0
FROM table
where _load_file_date >= '{{{{ ds }}}}'"""
)