An implementation of Scheduler that catches jobs that fail. For use with https://github.com/dbader/schedule

safe_schedule.py

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()

test_safe_schedule.py

#!/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)

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?

from datetime import timedelta, datetime

import schedule


class RetryableJob(schedule.Job):
    """
    A job that can be retried if failed.
    """

    def __init__(self, retry_after: timedelta = timedelta(minutes=30), cancel_after_consecutive_failures: int = 3,
                 *args, **kwargs):
        self.retry_after = retry_after
        self.cancel_after_consecutive_failures = cancel_after_consecutive_failures
        self._failure_count = 0
        super().__init__(*args, **kwargs)

    def mark_failed(self) -> None:
        self._failure_count += 1

    @property
    def should_retry(self) -> bool:
        return self._failure_count < self.cancel_after_consecutive_failures

    @property
    def prepare_next_run_time(self) -> datetime:
        return datetime.now() + self.retry_after


class SafeScheduler(schedule.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):
        super().__init__()

    def _run_job(self, job):
        try:
            super()._run_job(job)
            return
        except Exception as e:
            logging.error(e)
            if not isinstance(job, RetryableJob):
                logger.warn("Failed job cancelled")
                self.cancel_job(job)
                return

            job: RetryableJob = job  # type: ignore
            job.mark_failed()

            if job.should_retry:
                next_run_time = job.prepare_next_run_time
                logger.warn(f"Rescheduled failed retryable job for {next_run_time.isoformat()}")
                job.last_run = None
                job.next_run = next_run_time
                job._schedule_next_run()
                return
            else:
                logger.warn("Retryable job failed too many times, cancelled.")
                self.cancel_job(job)
                return

    def every_safe(self,
                   interval: int = 1,
                   /,
                   retry_after: timedelta = timedelta(minutes=30),
                   cancel_after_consecutive_failures: int = 3) -> RetryableJob:
        """
        Schedule a new periodic retryable job

        :param interval: A quantity of a certain time unit
        :param retry_after: An optional timedelta to use as the retry wait time
        :param cancel_after_consecutive_failures: An optional count of retry attempts before cancelling the job
        :return: An unconfigured :class:`RetryableJob <RetryableJob>`
        """

        return RetryableJob(
            retry_after=retry_after,
            cancel_after_consecutive_failures=cancel_after_consecutive_failures,
            interval=interval,
            scheduler=self
        )


scheduler = SafeScheduler()

def test_task():
    raise Exception("Retrigger the task")


scheduler.every_safe(3, retry_after=timedelta(minutes=31), cancel_after_consecutive_failures=3).seconds.do(test_task)

while True:
    scheduler.run_pending()
    time.sleep(1)