diff --git a/airflow/sensors/time_sensor.py b/airflow/sensors/time_sensor.py index 2ac93894eda57..72416be9e3231 100644 --- a/airflow/sensors/time_sensor.py +++ b/airflow/sensors/time_sensor.py @@ -15,6 +15,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +import datetime from airflow.sensors.base import BaseSensorOperator from airflow.triggers.temporal import DateTimeTrigger @@ -50,9 +51,15 @@ class TimeSensorAsync(BaseSensorOperator): def __init__(self, *, target_time, **kwargs): super().__init__(**kwargs) self.target_time = target_time + current_time = timezone.make_naive(timezone.utcnow(), self.dag.timezone) + todays_date = current_time.date() + self.target_datetime = datetime.datetime.combine(todays_date, self.target_time, current_time.tzinfo) def execute(self, context): - self.defer(trigger=DateTimeTrigger(moment=self.target_time), method_name="execute_complete") + self.defer( + trigger=DateTimeTrigger(moment=self.target_datetime), + method_name="execute_complete", + ) def execute_complete(self, context, event=None): # pylint: disable=unused-argument """Callback for when the trigger fires - returns immediately.""" diff --git a/airflow/triggers/base.py b/airflow/triggers/base.py index d48be6dccce49..a7bb44c206212 100644 --- a/airflow/triggers/base.py +++ b/airflow/triggers/base.py @@ -30,7 +30,7 @@ class BaseTrigger(abc.ABC): We use the same class for both situations, and rely on all Trigger classes to be able to return the (Airflow-JSON-encodable) arguments that will - let them be reinsantiated elsewhere. + let them be re-instantiated elsewhere. """ def __init__(self):