pankajastro commented on code in PR #76270:
URL: https://github.com/apache/airflow/pull/76270#discussion_r4452607458
##########
providers/common/sql/src/airflow/providers/common/sql/sensors/sql.py:
##########
@@ -17,15 +17,41 @@
from __future__ import annotations
from collections.abc import Callable, Mapping, Sequence
+from datetime import timedelta
from operator import itemgetter
from typing import TYPE_CHECKING, Any
-from airflow.providers.common.compat.sdk import AirflowException, BaseHook,
BaseSensorOperator
+from airflow.providers.common.compat.sdk import AirflowException, BaseHook,
BaseSensorOperator, conf
from airflow.providers.common.sql.hooks.sql import DbApiHook
+from airflow.providers.common.sql.triggers.sql import NO_ROWS_MESSAGE,
SqlSensorTrigger
+from airflow.providers.common.sql.version_compat import AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_2_2_PLUS
if TYPE_CHECKING:
from airflow.providers.common.compat.sdk import Context
+_first_cell = itemgetter(0)
+
+# Values that survive Airflow 2's JSON storage of trigger arguments unchanged.
+_JSON_SCALARS = (str, int, float, bool, type(None))
+
+
+def _is_first_cell(selector: Callable[[tuple[Any]], Any]) -> bool:
+ # Compare by value rather than identity: a deep copy of the sensor, or an
explicit
+ # ``selector=itemgetter(0)``, must still count as the default condition.
+ return isinstance(selector, itemgetter) and selector.__reduce__() ==
_first_cell.__reduce__()
+
+
+def _survives_json(value: Any) -> bool:
+ if isinstance(value, Mapping):
+ return all(isinstance(k, str) and _survives_json(v) for k, v in
value.items())
+ if isinstance(value, (list, tuple)):
+ return all(_survives_json(v) for v in value)
Review Comment:
Does Airflow 2.11 keep tuples when it stores trigger kwargs? If it turns
them into lists, a positional `parameters` tuple would reach the triggerer as a
list (psycopg2 renders a list as `ARRAY[...]`). If so, I'd drop `tuple` here so
the sensor pokes on the worker instead.
---
Drafted-by: Claude Code (Sonnet 5.5); reviewed by @pankajastro before posting
##########
providers/common/sql/docs/operators.rst:
##########
@@ -282,6 +282,53 @@ between two connections.
:start-after: [START howto_operator_generic_transfer]
:end-before: [END howto_operator_generic_transfer]
+.. _howto/sensor:SqlSensor:
+
+Wait for a SQL condition
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Use the :class:`~airflow.providers.common.sql.sensors.sql.SqlSensor` to wait
until a query returns a row
+whose first cell is truthy, for example until today's orders have landed. It
works with any connection
+whose hook is a :class:`~airflow.providers.common.sql.hooks.sql.DbApiHook`.
The examples below use
+PostgreSQL; write ``sql`` and ``parameters`` in your database's dialect and
placeholder style.
+
+.. exampleinclude:: /../tests/system/common/sql/example_sql_sensor.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_sensor_sql]
+ :end-before: [END howto_sensor_sql]
+
+For a condition you cannot write in SQL, pass ``success`` or ``failure``
callables. They receive the first
+cell of the first row, or whatever ``selector`` returns for that row.
``fail_on_empty=True`` fails the task
+when the query returns no rows instead of checking again.
+
+Deferrable mode
+---------------
+
+With ``deferrable=True``, the sensor checks once on the worker. If the
condition is not met, it defers and
+the triggerer runs the query every ``poke_interval`` seconds, so the task
holds no worker slot while it
+waits. ``timeout`` caps the whole wait.
Review Comment:
Nit: the deferral timeout starts when the task defers, so the total wait can
exceed `timeout` by the time the first poke took. "``timeout`` caps the time
spent deferred" would be more accurate.
---
Drafted-by: Claude Code (Sonnet 5.5); reviewed by @pankajastro before posting
--
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]
