Skip to content

fix(calendar): localize cron start_time to timetable timezone before computing planned runs #71248

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
waterWang wants to merge 1 commit into
base: main
Choose a base branch
from
+21 −6
Open

fix(calendar): localize cron start_time to timetable timezone before computing planned runs #71248

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 21 additions & 6 deletions airflow-core/src/airflow/api_fastapi/core_api/services/ui/calendar.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
from sqlalchemy.orm import InstrumentedAttribute, Session

from airflow._shared.timezones import timezone
from airflow._shared.timezones.timezone import convert_to_utc, make_aware, make_naive
from airflow.api_fastapi.common.parameters import RangeFilter
from airflow.api_fastapi.core_api.datamodels.ui.calendar import (
CalendarDeadlineCollectionResponse,
Expand Down Expand Up @@ -215,21 +216,35 @@ def _calculate_cron_planned_runs(
"""Calculate planned runs for cron-based timetables."""
dates: dict[datetime, int] = collections.Counter()

timetable = cast("CronMixin", dag.timetable)
tz = timetable._timezone

# Localize start_time into the timetable's own timezone before
# constructing croniter, matching the pattern used by
# CronMixin._get_next(). Without this, the cron expression is
# matched against UTC wall-clock, which produces wrong scheduled
# times for DAGs with a non-UTC default_timezone.
local_start = make_naive(last_data_interval.end, tz)

dates_iter: Iterator[datetime | None] = croniter(
cast("CronMixin", dag.timetable)._expression,
start_time=last_data_interval.end,
timetable._expression,
start_time=local_start,
ret_type=datetime,
)

for dt in dates_iter:
if dt is None or dt.year != year:
if dt is None:
break
# Convert back to UTC-aware for comparison and storage.
dt_utc = convert_to_utc(make_aware(dt, tz))
if dt_utc.year != year:
break
if dag.end_date and dt > dag.end_date:
if dag.end_date and dt_utc > dag.end_date:
break
if not self._is_date_in_range(dt, date_filter):
if not self._is_date_in_range(dt_utc, date_filter):
continue

dates[self._truncate_datetime_for_granularity(dt, granularity)] += 1
dates[self._truncate_datetime_for_granularity(dt_utc, granularity)] += 1

return [
CalendarTimeRangeResponse(date=dt, state="planned", count=count) for dt, count in dates.items()
Expand Down