Last active
December 15, 2023 13:09
-
-
Save mplewis/8483f1c24f2d6259aef6 to your computer and use it in GitHub Desktop.
An implementation of Scheduler that catches jobs that fail. For use with https://github.com/dbader/schedule
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
from traceback import format_exc | |
import datetime | |
from schedule import Scheduler | |
logger = logging.getLogger('schedule') | |
class SafeScheduler(Scheduler): | |
""" | |
An implementation of Scheduler that catches jobs that fail, logs their | |
exception tracebacks as errors, optionally reschedules the jobs for their | |
next run time, and keeps going. | |
Use this to run jobs that may or may not crash without worrying about | |
whether other jobs will run or if they'll crash the entire script. | |
""" | |
def __init__(self, reschedule_on_failure=True): | |
""" | |
If reschedule_on_failure is True, jobs will be rescheduled for their | |
next run as if they had completed successfully. If False, they'll run | |
on the next run_pending() tick. | |
""" | |
self.reschedule_on_failure = reschedule_on_failure | |
super().__init__() | |
def _run_job(self, job): | |
try: | |
super()._run_job(job) | |
except Exception: | |
logger.error(format_exc()) | |
job.last_run = datetime.datetime.now() | |
job._schedule_next_run() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
import time | |
from safe_schedule import SafeScheduler | |
def good_task_1(): | |
print('Good Task 1') | |
def good_task_2(): | |
print('Good Task 2') | |
def good_task_3(): | |
print('Good Task 3') | |
def bad_task_1(): | |
print('Bad Task 1') | |
print(1/0) | |
def bad_task_2(): | |
print('Bad Task 2') | |
raise Exception('Something went wrong!') | |
scheduler = SafeScheduler() | |
scheduler.every(3).seconds.do(good_task_1) | |
scheduler.every(5).seconds.do(bad_task_1) | |
scheduler.every(7).seconds.do(good_task_2) | |
scheduler.every(8).seconds.do(bad_task_2) | |
scheduler.every(12).seconds.do(good_task_3) | |
while True: | |
scheduler.run_pending() | |
time.sleep(1) |
Hi,
I modified your extension in a way that the job can be rescheduled in minutes o seconds after a failure, or can be even canceled on failure.
Thanks for the work.
Bye
class SafeScheduler(Scheduler):
"""
An implementation of Scheduler that catches jobs that fail, logs their
exception tracebacks as errors, optionally reschedules the jobs for their
next run time, and keeps going.
Use this to run jobs that may or may not crash without worrying about
whether other jobs will run or if they'll crash the entire script.
"""
def __init__(self, reschedule_on_failure=True, minutes_after_failure=0, seconds_after_failure=0):
"""
If reschedule_on_failure is True, jobs will be rescheduled for their
next run as if they had completed successfully. If False, they'll run
on the next run_pending() tick.
"""
self.reschedule_on_failure = reschedule_on_failure
self.minutes_after_failure = minutes_after_failure
self.seconds_after_failure = seconds_after_failure
super().__init__()
def _run_job(self, job):
try:
super()._run_job(job)
except Exception:
logger.error(format_exc())
if(self.reschedule_on_failure):
if(self.minutes_after_failure!=0 or self.seconds_after_failure!=0):
logger.warn("Rescheduled in %s minutes and %s seconds." % (self.minutes_after_failure, self.seconds_after_failure))
job.last_run = None
job.next_run = datetime.now() + timedelta(minutes=self.minutes_after_failure, seconds=self.seconds_after_failure)
else:
logger.warn("Rescheduled.")
job.last_run = datetime.now()
job._schedule_next_run()
else:
logger.warn("Job canceled.")
self.cancel_job(job)
I got the following error " AttributeError: module 'datetime' has no attribute 'now' ", so I changed datetime.now() to datetime.datetime.now().
@fimasini Thanks for the contributed
@mplewis Thanks for awesome initiative
class SafeScheduler(Scheduler):
"""
An implementation of Scheduler that catches jobs that fail, logs their
exception tracebacks as errors, optionally reschedules the jobs for their
next run time, and keeps going.
Use this to run jobs that may or may not crash without worrying about
whether other jobs will run or if they'll crash the entire script.
"""
def __init__(self, reschedule_on_failure=True, minutes_after_failure=0, seconds_after_failure=0):
"""
If reschedule_on_failure is True, jobs will be rescheduled for their
next run as if they had completed successfully. If False, they'll run
on the next run_pending() tick.
"""
self.reschedule_on_failure = reschedule_on_failure
self.minutes_after_failure = minutes_after_failure
self.seconds_after_failure = seconds_after_failure
super().__init__()
def _run_job(self, job):
try:
super()._run_job(job)
except Exception:
logger.error(format_exc())
if(self.reschedule_on_failure):
if(self.minutes_after_failure!=0 or self.seconds_after_failure!=0):
logger.warn("Rescheduled in %s minutes and %s seconds." % (self.minutes_after_failure, self.seconds_after_failure))
job.last_run = None
job.next_run = datetime.datetime.now() + timedelta(minutes=self.minutes_after_failure, seconds=self.seconds_after_failure)
else:
logger.warn("Rescheduled.")
job.last_run = datetime.datetime.now()
job._schedule_next_run()
else:
logger.warn("Job canceled.")
self.cancel_job(job)
Hello! I have noticed that if my job keeps failing too many times, the entire process gets killed because I get queue.Full
error. Any way to clear that up?
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
You're right, I don't think
reschedule_on_failure
does anything. I checked the parent class (version 0.6.0) and it doesn't reference that variable at all.