Add SKIP process in addition to occurring ERROR in SQLCheckOperator

[kotaaaa]

Description

In SQLCheckOperator,
I propose to add a feature of the check option that makes it optionally skip instead of occurring error if the condition meets.

Use case/motivation

I was looking for a feature similar to SQLCheckOperator.
Actually, in SQLCheckOperator, if it does not match the condition, it throws error. However, for my case, I wanted to skip rather than throwing error.

I asked for advise at slack about how I can implement this, then I followed @dstandish's advise so that I could develop for just my case. However, I want to contribute to community by adding a new feature to solve this.

Related issues

NONE

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[eladkal]

This feels more like a custom use case that you should solve locally in your deployment.
However, should this be something needed then this should be solved in BaseOprator.
What are the use cases for this request?

We have similar of soft_fail in sensors:

airflow/airflow/sensors/base.py

Line 83 in a3dd847

:param soft_fail: Set to true to mark the task as SKIPPED on failure

which allows all sensors to set to skip on timeout.

[kotaaaa]

@eladkal, thank you for your reply!
For my case, process is folked by the result of select count(*) from {bigquery's table}
so, I created class like below.
In not only my case, but also general usecase, it might happen that the result of the query to bigquery's table will decide whether it should skip the process or proceed, I thought.
(below is actually Airflow v1)

from airflow.exceptions import AirflowException
from airflow.exceptions import AirflowSkipException

from airflow.models import BaseOperator
from airflow.hooks.base_hook import BaseHook
from airflow.contrib.hooks.bigquery_hook import BigQueryHook
from airflow.utils.decorators import apply_defaults

class SQLCheckSkipOperator(BaseOperator):
    ...

    def execute(self, context=None):
        self.log.info("Executing SQL check: %s", self.sql)
        records = self.get_db_hook().get_first(self.sql)

        self.log.info("Record: %s", records)
        if not records:
            raise AirflowSkipException("The query returned None") # skip the upstream tasks
        elif not all([bool(r) for r in records]):
            raise AirflowSkipException(
                "Test failed.\nQuery:\n{query}\nResults:\n{records!s}".format(
                    query=self.sql, records=records
                )
            )

        self.log.info("Success.")
...

class BigQueryCheckSkipOperator(SQLCheckSkipOperator):
...