omarismail94 commented on issue #13750:
URL: https://github.com/apache/airflow/issues/13750#issuecomment-767026584
@mik-laj Yeah, you have to do something like this:
```
class BigQuerySqlSensor(BaseSensorOperator):
template_fields = ('sql',)
template_ext= ('.sql',)
@apply_defaults
def __init__(
self,
bigquery_conn_id = 'bigquery_conn_id',
delegate_to=None,
location='US',
sql=None,
use_legacy_sql=False,
**kwargs):
self.bigquery_conn_id = bigquery_conn_id
self.delegate_to = delegate_to
self.location = location
self.sql = sql
self.use_legacy_sql = use_legacy_sql
super().__init__(**kwargs)
def poke(self, context):
hook = BigQueryHook(bigquery_conn_id=self.bigquery_conn_id,
delegate_to=self.delegate_to, location=self.location, use_legacy_sql=
self.use_legacy_sql)
connection = hook.get_conn()
cursor = connection.cursor()
cursor.execute(self.sql)
for row in cursor.fetchall():
self.log.info("printing rows ...")
self.log.info(row)
row_count = row[0]
self.log.info("rows printed.")
if row_count > 0:
return True
return 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.
For queries about this service, please contact Infrastructure at:
[email protected]
