build.scheduler

Complexity

Total Complexity

6

Size/Duplication

Total Lines	44
Duplicated Lines	0 %

Test Coverage

Coverage

94.44%

Importance

Changes

0

5 Methods

Rating	Name	Duplication	Size	Complexity
A	Scheduler.get_job()	0	3	1
A	Scheduler.shutdown()	0	6	1
A	Scheduler.remove_job()	0	6	2
A	Scheduler.__init__()	0	4	1
A	Scheduler.add_callable()	0	6	1

"""scheduler used by the sndtrace_cp

This module is intended to replace the usage of the napp
called scheduler soon to be deprecated.
"""
import pytz
from apscheduler.jobstores.base import JobLookupError
from apscheduler.schedulers.background import BackgroundScheduler


class Scheduler:
    """Class to set scheduler."""

    def __init__(self) -> None:
        """Create an instance of Scheduler and starts it."""
        self.scheduler = BackgroundScheduler(timezone=pytz.utc)
        self.scheduler.start()

    # pylint: disable=too-many-arguments
    def add_callable(self, id_, func, trigger, args=None, kwargs=None,
                     **trigger_args):
        """Add job to scheduler"""
        return self.scheduler.add_job(func, trigger=trigger, args=args,
                                      kwargs=kwargs,
                                      id=id_, **trigger_args)

    def remove_job(self, id_):
        """Remove job from scheduler"""
        try:
            self.scheduler.remove_job(id_)
        except JobLookupError:
            pass

    def shutdown(self, wait):
        """Shut down scheduler.
        Delete the jobs inside the scheduler before calling
        this function.
        """
        self.scheduler.shutdown(wait)

    def get_job(self, id_):
        """Return an scheduled job with id_ as id"""
        return self.scheduler.get_job(id_)