Source code for irc.schedule

import abc

from tempora import schedule


class IScheduler(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def execute_every(self, period, func):
        "execute func every period"

    @abc.abstractmethod
    def execute_at(self, when, func):
        "execute func at when"

    @abc.abstractmethod
    def execute_after(self, delay, func):
        "execute func after delay"

    @abc.abstractmethod
    def run_pending(self):
        "invoke the functions that are due"


class DefaultScheduler(schedule.InvokeScheduler, IScheduler):
    def execute_every(self, period, func):
        """
        Executes `func` every `period`.

        :param `func`: function to execute
        :param `period`: `int` in seconds, or `datetime.timedelta`
        """
        self.add(schedule.PeriodicCommand.after(period, func))

    def execute_at(self, when, func):
        self.add(schedule.DelayedCommand.at_time(when, func))

    def execute_after(self, delay, func):
        self.add(schedule.DelayedCommand.after(delay, func))