Skip to content

Let _calc take a datetime object instead of a timestamp #168

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

Merged
potiuk merged 3 commits into from
Jun 22, 2025
Merged

Let _calc take a datetime object instead of a timestamp #168

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4a6eda9
Avoid changing now input parameter in _calc
bdrung Jun 21, 2025
0124f42
Calculate initial offset on datetime object in _calc
bdrung Jun 21, 2025
1a15e97
Let _calc take a datetime object instead of a timestamp
bdrung Jun 21, 2025
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
31 changes: 16 additions & 15 deletions src/croniter/croniter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -358,6 +358,7 @@ def _get_next(
raise TypeError("Invalid ret_type, only 'float' or 'datetime' is acceptable.")

result = self._calc_next(is_prev)
result = self.datetime_to_timestamp(result)

# DST Handling for cron job spanning across days
dtstarttime = self._timestamp_to_datetime(self.dst_start_time)
Expand Down Expand Up @@ -440,7 +441,8 @@ def __iter__(self):

__next__ = next = _get_next

def _calc_next(self, is_prev: bool) -> float:
def _calc_next(self, is_prev: bool) -> datetime.datetime:
current = self.timestamp_to_datetime(self.cur)
expanded = self.expanded[:]
nth_weekday_of_month = self.nth_weekday_of_month.copy()

Expand All @@ -458,33 +460,32 @@ def _calc_next(self, is_prev: bool) -> float:
else:
bak = expanded[DOW_FIELD]
expanded[DOW_FIELD] = ["*"]
t1 = self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
t1 = self._calc(current, expanded, nth_weekday_of_month, is_prev)
expanded[DOW_FIELD] = bak
expanded[DAY_FIELD] = ["*"]

t2 = self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
t2 = self._calc(current, expanded, nth_weekday_of_month, is_prev)
if is_prev:
return t1 if t1 > t2 else t2
return t1 if t1 < t2 else t2

return self._calc(self.cur, expanded, nth_weekday_of_month, is_prev)
return self._calc(current, expanded, nth_weekday_of_month, is_prev)

def _calc(self, now, expanded, nth_weekday_of_month, is_prev):
if is_prev:
now = math.ceil(now)
nearest_diff_method = self._get_prev_nearest_diff
sign = -1
offset = 1 if (len(expanded) > UNIX_CRON_LEN or now % 60 > 0) else 60
offset = relativedelta(microseconds=-1)
else:
now = math.floor(now)
nearest_diff_method = self._get_next_nearest_diff
sign = 1
offset = 1 if (len(expanded) > UNIX_CRON_LEN) else 60

dst = now = self.timestamp_to_datetime(now + sign * offset)
offset = relativedelta(seconds=1) if len(expanded) > UNIX_CRON_LEN else relativedelta(minutes=1)
dst = now + offset
if len(expanded) > UNIX_CRON_LEN:
dst = dst.replace(microsecond=0)
else:
dst = dst.replace(second=0, microsecond=0)

month, year = dst.month, dst.year
current_year = now.year
month = dst.month
year = current_year = dst.year

def proc_year(d):
if len(expanded) == YEAR_CRON_LEN:
Expand Down Expand Up @@ -677,7 +678,7 @@ def proc_second(d):
break
if next:
continue
return self.datetime_to_timestamp(dst.replace(microsecond=0))
return dst.replace(microsecond=0)

if is_prev:
raise CroniterBadDateError("failed to find prev date")
Expand Down