calibre-web/cps/services/background_scheduler.py

# -*- coding: utf-8 -*-

#   This file is part of the Calibre-Web (https://github.com/janeczku/calibre-web)
#     Copyright (C) 2020 mmonkey
#
#   This program is free software: you can redistribute it and/or modify
#   it under the terms of the GNU General Public License as published by
#   the Free Software Foundation, either version 3 of the License, or
#   (at your option) any later version.
#
#   This program is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU General Public License for more details.
#
#   You should have received a copy of the GNU General Public License
#   along with this program. If not, see <http://www.gnu.org/licenses/>.

import atexit

from .. import logger
from .worker import WorkerThread

try:
    from apscheduler.schedulers.background import BackgroundScheduler as BScheduler
    from apscheduler.triggers.cron import CronTrigger
    from apscheduler.triggers.date import DateTrigger
    use_APScheduler = True
except (ImportError, RuntimeError) as e:
    use_APScheduler = False
    log = logger.create()
    log.info('APScheduler not found. Unable to schedule tasks.')


class BackgroundScheduler:
    _instance = None

    def __new__(cls):
        if not use_APScheduler:
            return False

        if cls._instance is None:
            cls._instance = super(BackgroundScheduler, cls).__new__(cls)
            cls.log = logger.create()
            cls.scheduler = BScheduler()
            cls.scheduler.start()

        return cls._instance

    def schedule(self, func, trigger, name=None):
        if use_APScheduler:
            return self.scheduler.add_job(func=func, trigger=trigger, name=name)

    # Expects a lambda expression for the task
    def schedule_task(self, task, user=None, name=None, hidden=False, trigger=None):
        if use_APScheduler:
            def scheduled_task():
                worker_task = task()
                worker_task.scheduled = True
                WorkerThread.add(user, worker_task, hidden=hidden)
            return self.schedule(func=scheduled_task, trigger=trigger, name=name)

    # Expects a list of lambda expressions for the tasks
    def schedule_tasks(self, tasks, user=None, trigger=None):
        if use_APScheduler:
            for task in tasks:
                self.schedule_task(task[0], user=user, trigger=trigger, name=task[1], hidden=task[2])

    # Expects a lambda expression for the task
    def schedule_task_immediately(self, task, user=None, name=None, hidden=False):
        if use_APScheduler:
            def immediate_task():
                WorkerThread.add(user, task(), hidden)
            return self.schedule(func=immediate_task, trigger=DateTrigger(), name=name)

    # Expects a list of lambda expressions for the tasks
    def schedule_tasks_immediately(self, tasks, user=None):
        if use_APScheduler:
            for task in tasks:
                self.schedule_task_immediately(task[0], user, name="immediately " + task[1], hidden=task[2])

    # Remove all jobs
    def remove_all_jobs(self):
        self.scheduler.remove_all_jobs()