vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-04-04 01:01:35 -07:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Refactor ensure_future_datetime to simplify logic and remove max_retries parameter

Browse Source
This commit is contained in:
Jokob @NetAlertX
2026-03-04 03:07:37 +00:00
parent ba26f34191
commit c73ce839f2
1 changed files with 3 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
server/utils/datetime_utils.py
View File

@@ -115,24 +115,19 @@ def is_datetime_future(dt, current_threshold=None):
return dt > current_threshold return dt > current_threshold
def ensure_future_datetime(schedule_obj, current_threshold=None, max_retries=5): def ensure_future_datetime(schedule_obj, current_threshold=None):
""" """
Ensure a schedule's next() call returns a datetime strictly in the future. Ensure a schedule's next() call returns a datetime strictly in the future.
This is a defensive utility for cron/schedule libraries that should always return Keeps calling .next() until a future time is returned — never raises.
future times but may have edge cases. Validates and retries if needed.
Args: Args:
schedule_obj: A schedule object with a .next() method (e.g., from croniter/APScheduler) schedule_obj: A schedule object with a .next() method (e.g., from croniter/APScheduler)
current_threshold: datetime to compare against. If None, uses timeNowTZ(as_string=False) current_threshold: datetime to compare against. If None, uses timeNowTZ(as_string=False)
max_retries: Maximum times to call .next() if result is not in future (default: 5)
Returns: Returns:
datetime.datetime: A guaranteed future datetime from schedule_obj.next() datetime.datetime: A guaranteed future datetime from schedule_obj.next()
Raises:
RuntimeError: If max_retries exceeded without getting a future time
Examples: Examples:
newSchedule = Cron(run_sch).schedule(start_date=timeNowUTC(as_string=False)) newSchedule = Cron(run_sch).schedule(start_date=timeNowUTC(as_string=False))
next_time = ensure_future_datetime(newSchedule) next_time = ensure_future_datetime(newSchedule)
@@ -141,17 +136,9 @@ def ensure_future_datetime(schedule_obj, current_threshold=None, max_retries=5):
current_threshold = timeNowTZ(as_string=False) current_threshold = timeNowTZ(as_string=False)
next_time = schedule_obj.next() next_time = schedule_obj.next()
retries = 0
while next_time <= current_threshold and retries < max_retries: while next_time <= current_threshold:
next_time = schedule_obj.next() next_time = schedule_obj.next()
retries += 1
if next_time <= current_threshold:
raise RuntimeError(
f"[ensure_future_datetime] Failed to get future time after {max_retries} retries. "
f"Last attempt: {next_time}, Current time: {current_threshold}"
)
return next_time return next_time