From 791fc8dc6ff686b992dea88506e6f9bc5a1a0f8d Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 23 Jul 2026 11:39:20 -0700 Subject: [PATCH 1/4] Fix DateTimeSensor crash when target_time renders to a datetime target_time is a template field, so it is only rendered after __init__ runs. DateTimeSensor validated and normalized it in the constructor (acting on the un-rendered Jinja expression), and poke() parsed self.target_time directly with timezone.parse(). With a templated target_time and render_template_as_native_obj enabled, the field renders to a datetime and poke() raised "TypeError: 'DateTime' object cannot be cast as 'str'". Store target_time verbatim in the constructor and defer normalization and validation to _moment, the accessor that already runs after rendering; poke() now goes through it. related: #70296 Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../providers/standard/sensors/date_time.py | 32 ++++----- .../unit/standard/sensors/test_date_time.py | 72 ++++++++++++------- .../validate_operators_init_exemptions.txt | 3 +- 3 files changed, 61 insertions(+), 46 deletions(-) diff --git a/providers/standard/src/airflow/providers/standard/sensors/date_time.py b/providers/standard/src/airflow/providers/standard/sensors/date_time.py index e0b64fbf51838..ba6abdf7cea2c 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/date_time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/date_time.py @@ -63,30 +63,24 @@ class DateTimeSensor(BaseSensorOperator): def __init__(self, *, target_time: str | datetime.datetime, **kwargs) -> None: super().__init__(**kwargs) - - # self.target_time can't be a datetime object as it is a template_field - if isinstance(target_time, datetime.datetime): - self.target_time = target_time.isoformat() - elif isinstance(target_time, str): - self.target_time = target_time - else: - raise TypeError( - f"Expected str or datetime.datetime type for target_time. Got {type(target_time)}" - ) + # target_time is a template field, so it is rendered after __init__ runs. Store it + # verbatim and defer validation/normalization to _moment, which sees the rendered value. + self.target_time = target_time def poke(self, context: Context) -> bool: self.log.info("Checking if the time (%s) has come", self.target_time) - return timezone.utcnow() > timezone.parse(self.target_time) + return timezone.utcnow() > self._moment @property def _moment(self) -> datetime.datetime: - # Note following is reachable code if Jinja is used for redering template fields and - # render_template_as_native_obj=True is used. - # In this case, the target_time is already a datetime object. - if isinstance(self.target_time, datetime.datetime): # type:ignore[unreachable] - return self.target_time # type:ignore[unreachable] - - return timezone.parse(self.target_time) + # target_time is a template field: after rendering it is usually a string, but with + # render_template_as_native_obj=True it can already be a datetime. + target_time: Any = self.target_time + if isinstance(target_time, datetime.datetime): + return target_time + if isinstance(target_time, str): + return timezone.parse(target_time) + raise TypeError(f"Expected str or datetime.datetime type for target_time. Got {type(target_time)}") class DateTimeSensorAsync(DateTimeSensor): @@ -126,7 +120,7 @@ def __init__( self.start_from_trigger = start_from_trigger if self.start_from_trigger: self.start_trigger_args.trigger_kwargs = dict( - moment=timezone.parse(self.target_time), + moment=self._moment, end_from_trigger=self.end_from_trigger, ) diff --git a/providers/standard/tests/unit/standard/sensors/test_date_time.py b/providers/standard/tests/unit/standard/sensors/test_date_time.py index 6b07eae4948ec..2e5c0ab4d26cb 100644 --- a/providers/standard/tests/unit/standard/sensors/test_date_time.py +++ b/providers/standard/tests/unit/standard/sensors/test_date_time.py @@ -17,6 +17,7 @@ # under the License. from __future__ import annotations +import datetime from unittest.mock import patch import pendulum @@ -24,7 +25,7 @@ from airflow import macros from airflow.models.dag import DAG -from airflow.providers.standard.sensors.date_time import DateTimeSensor +from airflow.providers.standard.sensors.date_time import DateTimeSensor, DateTimeSensorAsync from tests_common.test_utils.version_compat import timezone @@ -38,41 +39,32 @@ def setup_class(cls): cls.dag = DAG("test_dag", schedule=None, default_args=args) @pytest.mark.parametrize( - ("task_id", "target_time", "expected"), + ("task_id", "target_time"), [ - ( - "valid_datetime", - timezone.datetime(2020, 7, 6, 13, tzinfo=timezone.utc), - "2020-07-06T13:00:00+00:00", - ), - ( - "valid_str", - "20200706T210000+8", - "20200706T210000+8", - ), - ( - "jinja_str_is_accepted", - "{{ ds }}", - "{{ ds }}", - ), + ("valid_datetime", timezone.datetime(2020, 7, 6, 13, tzinfo=timezone.utc)), + ("valid_str", "20200706T210000+8"), + ("jinja_str_is_accepted", "{{ ds }}"), ], ) - def test_valid_input(self, task_id, target_time, expected): - """target_time should be a string as it is a template field""" + def test_target_time_stored_verbatim(self, task_id, target_time): + """target_time is a template field, so __init__ must store it as-is without transformation.""" op = DateTimeSensor( task_id=task_id, target_time=target_time, dag=self.dag, ) - assert op.target_time == expected + assert op.target_time == target_time - def test_invalid_input(self): + def test_invalid_input_rejected_after_rendering(self): + # __init__ no longer validates the template field; the TypeError surfaces when the + # un-renderable value is used at poke time. + op = DateTimeSensor( + task_id="test", + target_time=timezone.utcnow().time(), + dag=self.dag, + ) with pytest.raises(TypeError): - DateTimeSensor( - task_id="test", - target_time=timezone.utcnow().time(), - dag=self.dag, - ) + op.poke(None) @pytest.mark.parametrize( ("task_id", "target_time", "expected"), @@ -124,3 +116,31 @@ def test_moment(self, native, target_time, expected_type): sensor.render_template_fields(ctx) assert isinstance(sensor._moment, expected_type) + + @patch( + "airflow.providers.standard.sensors.date_time.timezone.utcnow", + return_value=pendulum.datetime(2020, 1, 2, tz="UTC"), + ) + def test_poke_with_natively_rendered_datetime(self, mock_utcnow): + """poke must handle a target_time rendered to a datetime (render_template_as_native_obj=True).""" + dag = DAG( + dag_id="native_poke_dag", + start_date=pendulum.datetime(2025, 1, 1, tz="UTC"), + schedule=None, + render_template_as_native_obj=True, + ) + op = DateTimeSensor(task_id="native_poke", target_time="{{ data_interval_end }}", dag=dag) + op.render_template_fields( + {"data_interval_end": pendulum.datetime(2020, 1, 1, tz="UTC"), "macros": macros, "dag": dag} + ) + assert isinstance(op.target_time, datetime.datetime) + assert op.poke(None) is True + + def test_async_start_from_trigger_moment(self): + op = DateTimeSensorAsync( + task_id="async", + target_time="2020-01-01T00:00:00+00:00", + start_from_trigger=True, + dag=self.dag, + ) + assert op.start_trigger_args.trigger_kwargs["moment"] == pendulum.parse("2020-01-01T00:00:00+00:00") diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index b78123db0dc66..e22d6550a1150 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -68,4 +68,5 @@ providers/ssh/src/airflow/providers/ssh/operators/ssh.py::SSHOperator providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py::SSHRemoteJobOperator providers/standard/src/airflow/providers/standard/operators/bash.py::BashOperator providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py::TriggerDagRunOperator -providers/standard/src/airflow/providers/standard/sensors/date_time.py::DateTimeSensor +providers/teradata/src/airflow/providers/teradata/transfers/teradata_to_teradata.py::TeradataToTeradataOperator +providers/weaviate/src/airflow/providers/weaviate/operators/weaviate.py::WeaviateIngestOperator From 9b0864a568c4efc2930cc4394984670502c29bf5 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 23 Jul 2026 23:02:13 -0700 Subject: [PATCH 2/4] Normalize DateTimeSensor target_time in poke instead of dropping it The earlier refactor removed the datetime-to-ISO-string normalization that the constructor used to apply to target_time, leaving a datetime stored in the template field. Relocate that normalization to poke (the post-render execute path) so a datetime target_time is again stored as an ISO string, while __init__ keeps only the plain assignment. Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../providers/standard/sensors/date_time.py | 10 ++++++---- .../unit/standard/sensors/test_date_time.py | 16 +++++++++++++--- 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/providers/standard/src/airflow/providers/standard/sensors/date_time.py b/providers/standard/src/airflow/providers/standard/sensors/date_time.py index ba6abdf7cea2c..77f63163fa614 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/date_time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/date_time.py @@ -63,18 +63,20 @@ class DateTimeSensor(BaseSensorOperator): def __init__(self, *, target_time: str | datetime.datetime, **kwargs) -> None: super().__init__(**kwargs) - # target_time is a template field, so it is rendered after __init__ runs. Store it - # verbatim and defer validation/normalization to _moment, which sees the rendered value. + # target_time is a template field; store it verbatim and normalize once rendered. self.target_time = target_time def poke(self, context: Context) -> bool: + # Normalize a rendered datetime to an ISO string here (moved from __init__, which only + # saw the un-rendered Jinja) so the stored template field stays a string. + if isinstance(self.target_time, datetime.datetime): + self.target_time = self.target_time.isoformat() self.log.info("Checking if the time (%s) has come", self.target_time) return timezone.utcnow() > self._moment @property def _moment(self) -> datetime.datetime: - # target_time is a template field: after rendering it is usually a string, but with - # render_template_as_native_obj=True it can already be a datetime. + # After rendering target_time is usually a str; native rendering can yield a datetime. target_time: Any = self.target_time if isinstance(target_time, datetime.datetime): return target_time diff --git a/providers/standard/tests/unit/standard/sensors/test_date_time.py b/providers/standard/tests/unit/standard/sensors/test_date_time.py index 2e5c0ab4d26cb..2491a3ba1410e 100644 --- a/providers/standard/tests/unit/standard/sensors/test_date_time.py +++ b/providers/standard/tests/unit/standard/sensors/test_date_time.py @@ -56,8 +56,6 @@ def test_target_time_stored_verbatim(self, task_id, target_time): assert op.target_time == target_time def test_invalid_input_rejected_after_rendering(self): - # __init__ no longer validates the template field; the TypeError surfaces when the - # un-renderable value is used at poke time. op = DateTimeSensor( task_id="test", target_time=timezone.utcnow().time(), @@ -122,7 +120,7 @@ def test_moment(self, native, target_time, expected_type): return_value=pendulum.datetime(2020, 1, 2, tz="UTC"), ) def test_poke_with_natively_rendered_datetime(self, mock_utcnow): - """poke must handle a target_time rendered to a datetime (render_template_as_native_obj=True).""" + """poke handles a target_time rendered to a datetime and normalizes it to an ISO string.""" dag = DAG( dag_id="native_poke_dag", start_date=pendulum.datetime(2025, 1, 1, tz="UTC"), @@ -135,6 +133,18 @@ def test_poke_with_natively_rendered_datetime(self, mock_utcnow): ) assert isinstance(op.target_time, datetime.datetime) assert op.poke(None) is True + assert op.target_time == "2020-01-01T00:00:00+00:00" + + @patch( + "airflow.providers.standard.sensors.date_time.timezone.utcnow", + return_value=timezone.datetime(2020, 1, 2, tzinfo=timezone.utc), + ) + def test_poke_normalizes_datetime_target_time(self, mock_utcnow): + """A datetime target_time is normalized to its ISO string at poke, as __init__ used to do.""" + target = timezone.datetime(2020, 1, 1, tzinfo=timezone.utc) + op = DateTimeSensor(task_id="normalize", target_time=target, dag=self.dag) + assert op.poke(None) is True + assert op.target_time == target.isoformat() def test_async_start_from_trigger_moment(self): op = DateTimeSensorAsync( From 0e4bc6a3de35cf8cbd3adc75f90c9e72711259d8 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Fri, 24 Jul 2026 10:56:44 -0700 Subject: [PATCH 3/4] Localize DateTimeSensor target_time in _moment so async path is covered poke() is not reached by DateTimeSensorAsync, which overrides execute(); moving the datetime normalization into _moment (the shared post-render accessor) localizes a naive datetime for both the sync and async sensors, and drops the explanatory comments. Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../providers/standard/sensors/date_time.py | 8 +----- .../unit/standard/sensors/test_date_time.py | 27 ++++++++++--------- 2 files changed, 16 insertions(+), 19 deletions(-) diff --git a/providers/standard/src/airflow/providers/standard/sensors/date_time.py b/providers/standard/src/airflow/providers/standard/sensors/date_time.py index 77f63163fa614..d000ef1cd8b90 100644 --- a/providers/standard/src/airflow/providers/standard/sensors/date_time.py +++ b/providers/standard/src/airflow/providers/standard/sensors/date_time.py @@ -63,23 +63,17 @@ class DateTimeSensor(BaseSensorOperator): def __init__(self, *, target_time: str | datetime.datetime, **kwargs) -> None: super().__init__(**kwargs) - # target_time is a template field; store it verbatim and normalize once rendered. self.target_time = target_time def poke(self, context: Context) -> bool: - # Normalize a rendered datetime to an ISO string here (moved from __init__, which only - # saw the un-rendered Jinja) so the stored template field stays a string. - if isinstance(self.target_time, datetime.datetime): - self.target_time = self.target_time.isoformat() self.log.info("Checking if the time (%s) has come", self.target_time) return timezone.utcnow() > self._moment @property def _moment(self) -> datetime.datetime: - # After rendering target_time is usually a str; native rendering can yield a datetime. target_time: Any = self.target_time if isinstance(target_time, datetime.datetime): - return target_time + target_time = target_time.isoformat() if isinstance(target_time, str): return timezone.parse(target_time) raise TypeError(f"Expected str or datetime.datetime type for target_time. Got {type(target_time)}") diff --git a/providers/standard/tests/unit/standard/sensors/test_date_time.py b/providers/standard/tests/unit/standard/sensors/test_date_time.py index 2491a3ba1410e..96d125750cedc 100644 --- a/providers/standard/tests/unit/standard/sensors/test_date_time.py +++ b/providers/standard/tests/unit/standard/sensors/test_date_time.py @@ -120,7 +120,7 @@ def test_moment(self, native, target_time, expected_type): return_value=pendulum.datetime(2020, 1, 2, tz="UTC"), ) def test_poke_with_natively_rendered_datetime(self, mock_utcnow): - """poke handles a target_time rendered to a datetime and normalizes it to an ISO string.""" + """poke handles a target_time that native rendering resolved to a datetime.""" dag = DAG( dag_id="native_poke_dag", start_date=pendulum.datetime(2025, 1, 1, tz="UTC"), @@ -133,18 +133,11 @@ def test_poke_with_natively_rendered_datetime(self, mock_utcnow): ) assert isinstance(op.target_time, datetime.datetime) assert op.poke(None) is True - assert op.target_time == "2020-01-01T00:00:00+00:00" - @patch( - "airflow.providers.standard.sensors.date_time.timezone.utcnow", - return_value=timezone.datetime(2020, 1, 2, tzinfo=timezone.utc), - ) - def test_poke_normalizes_datetime_target_time(self, mock_utcnow): - """A datetime target_time is normalized to its ISO string at poke, as __init__ used to do.""" - target = timezone.datetime(2020, 1, 1, tzinfo=timezone.utc) - op = DateTimeSensor(task_id="normalize", target_time=target, dag=self.dag) - assert op.poke(None) is True - assert op.target_time == target.isoformat() + def test_moment_localizes_naive_datetime(self): + """A naive datetime target_time is localized to UTC via _moment (mirrors old isoformat()).""" + op = DateTimeSensor(task_id="naive", target_time=datetime.datetime(2020, 1, 1), dag=self.dag) + assert op._moment == pendulum.datetime(2020, 1, 1, tz="UTC") def test_async_start_from_trigger_moment(self): op = DateTimeSensorAsync( @@ -154,3 +147,13 @@ def test_async_start_from_trigger_moment(self): dag=self.dag, ) assert op.start_trigger_args.trigger_kwargs["moment"] == pendulum.parse("2020-01-01T00:00:00+00:00") + + def test_async_start_from_trigger_localizes_naive_datetime(self): + """DateTimeSensorAsync never pokes, so _moment must still localize a naive datetime.""" + op = DateTimeSensorAsync( + task_id="async_naive", + target_time=datetime.datetime(2020, 1, 1), + start_from_trigger=True, + dag=self.dag, + ) + assert op.start_trigger_args.trigger_kwargs["moment"] == pendulum.datetime(2020, 1, 1, tz="UTC") From 2ffad8d2af254b0ded87a9a35f213d42811089ca Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Fri, 24 Jul 2026 12:26:43 -0700 Subject: [PATCH 4/4] Sync operator __init__ exemptions with main Drop exemption entries that sibling burn-down PRs already resolved on main, which a stale rebase base had re-introduced. Net change to this file is only the DateTimeSensor removal. Signed-off-by: 1fanwang <1fannnw@gmail.com> --- scripts/ci/prek/validate_operators_init_exemptions.txt | 2 -- 1 file changed, 2 deletions(-) diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index e22d6550a1150..c32e54555d09b 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -68,5 +68,3 @@ providers/ssh/src/airflow/providers/ssh/operators/ssh.py::SSHOperator providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py::SSHRemoteJobOperator providers/standard/src/airflow/providers/standard/operators/bash.py::BashOperator providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py::TriggerDagRunOperator -providers/teradata/src/airflow/providers/teradata/transfers/teradata_to_teradata.py::TeradataToTeradataOperator -providers/weaviate/src/airflow/providers/weaviate/operators/weaviate.py::WeaviateIngestOperator