Vamsi-klu commented on code in PR #69746:
URL: https://github.com/apache/airflow/pull/69746#discussion_r3688897295
##########
providers/standard/src/airflow/providers/standard/sensors/time.py:
##########
@@ -43,53 +47,50 @@ class TimeSensor(BaseSensorOperator):
"""
- start_trigger_args = StartTriggerArgs(
-
trigger_cls="airflow.providers.standard.triggers.temporal.DateTimeTrigger",
- trigger_kwargs={"moment": "", "end_from_trigger": False},
- next_method="execute_complete",
- next_kwargs=None,
- timeout=None,
- )
- start_from_trigger = False
-
def __init__(
self,
*,
target_time: datetime.time,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
- start_from_trigger: bool = False,
end_from_trigger: bool = False,
trigger_kwargs: dict[str, Any] | None = None,
**kwargs,
) -> None:
+ start_from_trigger = kwargs.pop("start_from_trigger", None)
Review Comment:
Agreed — start_from_trigger is back in the explicit __init__ signature so
inspect/IDE/docs see it and typos fail loudly.
##########
providers/standard/src/airflow/providers/standard/sensors/time.py:
##########
@@ -43,53 +47,50 @@ class TimeSensor(BaseSensorOperator):
"""
- start_trigger_args = StartTriggerArgs(
-
trigger_cls="airflow.providers.standard.triggers.temporal.DateTimeTrigger",
- trigger_kwargs={"moment": "", "end_from_trigger": False},
- next_method="execute_complete",
- next_kwargs=None,
- timeout=None,
- )
- start_from_trigger = False
-
def __init__(
self,
*,
target_time: datetime.time,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
- start_from_trigger: bool = False,
end_from_trigger: bool = False,
trigger_kwargs: dict[str, Any] | None = None,
**kwargs,
) -> None:
+ start_from_trigger = kwargs.pop("start_from_trigger", None)
+ if start_from_trigger is not None:
+ warnings.warn(
+ "start_from_trigger is deprecated and no longer supported for
TimeSensor. "
+ "It has been ignored. Target time is now always evaluated at
execution time. "
+ "Use deferrable=True to defer the sensor instead.",
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
super().__init__(**kwargs)
+ self.target_time = target_time
+ self.deferrable = deferrable
+ self.end_from_trigger = end_from_trigger
+ # Accepted for compatibility only; storing it would reintroduce
serialized Dag hash churn.
+ del trigger_kwargs
- # Create a "date-aware" timestamp that will be used as the
"target_datetime". This is a requirement
- # of the DateTimeTrigger
-
- # Get date considering dag.timezone
+ def _get_target_datetime(self) -> datetime.datetime:
+ """Compute target datetime at execution time, not parse time."""
+ dag_timezone = getattr(getattr(self, "dag", None), "timezone", None)
or timezone.utc
+ now_date = datetime.datetime.now(dag_timezone).date()
aware_time = timezone.coerce_datetime(
- datetime.datetime.combine(
- datetime.datetime.now(self.dag.timezone), target_time,
self.dag.timezone
- )
+ datetime.datetime.combine(now_date, self.target_time, dag_timezone)
)
+ return timezone.convert_to_utc(aware_time)
- # Now that the dag's timezone has made the datetime timezone aware, we
need to convert to UTC
- self.target_datetime = timezone.convert_to_utc(aware_time)
- self.deferrable = deferrable
- self.start_from_trigger = start_from_trigger
- self.end_from_trigger = end_from_trigger
-
- if self.start_from_trigger:
- self.start_trigger_args.trigger_kwargs = dict(
- moment=self.target_datetime,
end_from_trigger=self.end_from_trigger
- )
+ @property
+ def target_datetime(self) -> datetime.datetime:
Review Comment:
Updated: target_datetime is now resolved once and cached for the life of the
instance (per attempt). Docstring states that explicitly — two reads either
side of midnight return the same date for a given attempt; a fresh task
instance re-resolves against today.
##########
providers/standard/src/airflow/providers/standard/sensors/time.py:
##########
@@ -43,53 +47,50 @@ class TimeSensor(BaseSensorOperator):
"""
- start_trigger_args = StartTriggerArgs(
-
trigger_cls="airflow.providers.standard.triggers.temporal.DateTimeTrigger",
- trigger_kwargs={"moment": "", "end_from_trigger": False},
- next_method="execute_complete",
- next_kwargs=None,
- timeout=None,
- )
- start_from_trigger = False
-
def __init__(
self,
*,
target_time: datetime.time,
deferrable: bool = conf.getboolean("operators", "default_deferrable",
fallback=False),
- start_from_trigger: bool = False,
end_from_trigger: bool = False,
trigger_kwargs: dict[str, Any] | None = None,
**kwargs,
) -> None:
+ start_from_trigger = kwargs.pop("start_from_trigger", None)
+ if start_from_trigger is not None:
+ warnings.warn(
+ "start_from_trigger is deprecated and no longer supported for
TimeSensor. "
+ "It has been ignored. Target time is now always evaluated at
execution time. "
+ "Use deferrable=True to defer the sensor instead.",
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
super().__init__(**kwargs)
+ self.target_time = target_time
+ self.deferrable = deferrable
+ self.end_from_trigger = end_from_trigger
+ # Accepted for compatibility only; storing it would reintroduce
serialized Dag hash churn.
+ del trigger_kwargs
- # Create a "date-aware" timestamp that will be used as the
"target_datetime". This is a requirement
- # of the DateTimeTrigger
-
- # Get date considering dag.timezone
+ def _get_target_datetime(self) -> datetime.datetime:
+ """Compute target datetime at execution time, not parse time."""
+ dag_timezone = getattr(getattr(self, "dag", None), "timezone", None)
or timezone.utc
Review Comment:
Dropped the broad except Exception. Uses self._dag (no raise) so unattached
sensors fall back to UTC — needed for unit tests and construct-before-add_task.
When a Dag is present its timezone is used. Comment in
_serializable_dag_timezone / _resolve_target_datetime names that case.
--
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]
