Merge branch 'dbschedule'

celery/beat.py

@@ -7,14 +7,16 @@ import time
 import shelve
 import threading
 import multiprocessing
-from datetime import datetime
+from datetime import datetime, timedelta
 from UserDict import UserDict
 
 from celery import log
 from celery import conf
-from celery import registry as _registry
 from celery import platform
+from celery.execute import send_task
+from celery.schedules import schedule
 from celery.messaging import establish_connection
+from celery.utils import instantiate
 from celery.utils.info import humanize_seconds
 
 
@@ -25,13 +27,28 @@ class SchedulingError(Exception):
 class ScheduleEntry(object):
     """An entry in the scheduler.
 
-    :param task: see :attr:`task`.
+    :param name: see :attr:`name`.
+    :param schedule: see :attr:`schedule`.
+    :param args: see :attr:`args`.
+    :param kwargs: see :attr:`kwargs`.
     :keyword last_run_at: see :attr:`last_run_at`.
     :keyword total_run_count: see :attr:`total_run_count`.
 
-    .. attribute:: task
+    .. attribute:: name
 
-        The task class.
+        The task name.
+
+    .. attribute:: schedule
+
+        The schedule (run_every/crontab)
+
+    .. attribute:: args
+
+        Args to apply.
+
+    .. attribute:: kwargs
+
+        Keyword arguments to apply.
 
     .. attribute:: last_run_at
 
@@ -43,8 +60,13 @@ class ScheduleEntry(object):
 
     """
 
-    def __init__(self, name, last_run_at=None, total_run_count=None):
+    def __init__(self, name, schedule, args=(), kwargs={},
+            options={}, last_run_at=None, total_run_count=None):
         self.name = name
+        self.schedule = schedule
+        self.args = args
+        self.kwargs = kwargs
+        self.options = options
         self.last_run_at = last_run_at or datetime.now()
         self.total_run_count = total_run_count or 0
 
@@ -52,26 +74,31 @@ class ScheduleEntry(object):
         """Returns a new instance of the same class, but with
         its date and count fields updated."""
         return self.__class__(self.name,
+                              self.schedule,
+                              self.args,
+                              self.kwargs,
+                              self.options,
                               datetime.now(),
                               self.total_run_count + 1)
 
-    def is_due(self, task):
+    def is_due(self):
         """See :meth:`celery.task.base.PeriodicTask.is_due`."""
-        return task.is_due(self.last_run_at)
+        return self.schedule.is_due(self.last_run_at)
+
+    def __repr__(self):
+        return "<Entry: %s(*%s, **%s) {%s}>" % (self.name,
+                                                self.args,
+                                                self.kwargs,
+                                                self.schedule)
 
 
 class Scheduler(UserDict):
     """Scheduler for periodic tasks.
 
-    :keyword registry: see :attr:`registry`.
     :keyword schedule: see :attr:`schedule`.
     :keyword logger:  see :attr:`logger`.
     :keyword max_interval: see :attr:`max_interval`.
 
-    .. attribute:: registry
-
-        The task registry to use.
-
     .. attribute:: schedule
 
         The schedule dict/shelve.
@@ -85,10 +112,10 @@ class Scheduler(UserDict):
         Maximum time to sleep between re-checking the schedule.
 
     """
+    Entry = ScheduleEntry
 
-    def __init__(self, registry=None, schedule=None, logger=None,
+    def __init__(self, schedule=None, logger=None,
             max_interval=None):
-        self.registry = registry or _registry.TaskRegistry()
         self.data = schedule
         if self.data is None:
             self.data = {}
@@ -96,7 +123,10 @@ class Scheduler(UserDict):
         self.max_interval = max_interval or conf.CELERYBEAT_MAX_LOOP_INTERVAL
 
         self.cleanup()
-        self.schedule_registry()
+        self.setup_schedule()
+
+    def iterentries(self):
+        return self.schedule.itervalues()
 
     def tick(self):
         """Run a tick, that is one iteration of the scheduler.
@@ -107,16 +137,18 @@ class Scheduler(UserDict):
         remaining_times = []
         connection = establish_connection()
         try:
-            for entry in self.schedule.values():
-                is_due, next_time_to_run = self.is_due(entry)
+            for entry in self.iterentries():
+                is_due, next_time_to_run = entry.is_due()
                 if is_due:
                     debug("Scheduler: Sending due task %s" % entry.name)
                     try:
-                        result = self.apply_async(entry, connection=connection)
+                        result = self.apply_async(entry,
+                                      connection=connection)
                     except SchedulingError, exc:
                         error("Scheduler: %s" % exc)
                     else:
-                        debug("%s sent. id->%s" % (entry.name, result.task_id))
+                        debug("%s sent. id->%s" % (entry.name,
+                                                   result.task_id))
                 if next_time_to_run:
                     remaining_times.append(next_time_to_run)
         finally:
@@ -124,39 +156,47 @@ class Scheduler(UserDict):
 
         return min(remaining_times + [self.max_interval])
 
-    def get_task(self, name):
-        return self.registry[name]
-
-    def is_due(self, entry):
-        return entry.is_due(self.get_task(entry.name))
+    def reserve(self, entry):
+        new_entry = self.schedule[entry.name] = entry.next()
+        return new_entry
 
     def apply_async(self, entry, **kwargs):
-
         # Update timestamps and run counts before we actually execute,
         # so we have that done if an exception is raised (doesn't schedule
         # forever.)
-        entry = self.schedule[entry.name] = entry.next()
-        task = self.get_task(entry.name)
+        entry = self.reserve(entry)
+
+        print("APPLYING: %s" % (entry, ))
 
         try:
-            result = task.apply_async(**kwargs)
+            result = send_task(entry.name, entry.args, entry.kwargs,
+                               **entry.options)
         except Exception, exc:
             raise SchedulingError("Couldn't apply scheduled task %s: %s" % (
-                    task.name, exc))
+                    entry.name, exc))
         return result
 
-    def schedule_registry(self):
-        """Add the current contents of the registry to the schedule."""
-        for name, task in self.registry.periodic().items():
-            if name not in self.schedule:
-                self.logger.debug("Scheduler: "
-                    "Added periodic task %s to schedule" % name)
-            self.schedule.setdefault(name, ScheduleEntry(task.name))
+    def maybe_schedule(self, s, relative=False):
+        if isinstance(s, int):
+            return timedelta(seconds=s)
+        if isinstance(s, timedelta):
+            return schedule(s, relative)
+        return s
+
+    def setup_schedule(self):
+        self.data = self.dict_to_entries(conf.CELERYBEAT_SCHEDULE)
+
+    def dict_to_entries(self, dict_):
+        entries = {}
+        for name, entry in dict_.items():
+            relative = entry.pop("relative", None)
+            entry["schedule"] = self.maybe_schedule(entry["schedule"],
+                                                    relative)
+            entries[name] = self.Entry(**entry)
+        return entries
 
     def cleanup(self):
-        for task_name, entry in self.schedule.items():
-            if task_name not in self.registry:
-                self.schedule.pop(task_name, None)
+        pass
 
     @property
     def schedule(self):
@@ -165,30 +205,29 @@ class Scheduler(UserDict):
 
 class ClockService(object):
     scheduler_cls = Scheduler
-    registry = _registry.tasks
     open_schedule = lambda self, filename: shelve.open(filename)
 
     def __init__(self, logger=None,
             max_interval=conf.CELERYBEAT_MAX_LOOP_INTERVAL,
-            schedule_filename=conf.CELERYBEAT_SCHEDULE_FILENAME):
+            schedule=conf.CELERYBEAT_SCHEDULE,
+            schedule_filename=conf.CELERYBEAT_SCHEDULE_FILENAME,
+            scheduler_cls=None):
         self.logger = logger or log.get_default_logger()
         self.max_interval = max_interval
-        self.schedule_filename = schedule_filename
+        self.scheduler_cls = scheduler_cls or self.scheduler_cls
         self._shutdown = threading.Event()
         self._stopped = threading.Event()
-        self._schedule = None
+        self.schedule = schedule
         self._scheduler = None
-        self._in_sync = False
         silence = self.max_interval < 60 and 10 or 1
         self.debug = log.SilenceRepeated(self.logger.debug,
                                          max_iterations=silence)
 
     def start(self, embedded_process=False):
-        self.logger.info("ClockService: Starting...")
-        self.logger.debug("ClockService: "
-            "Ticking with max interval->%s, schedule->%s" % (
-                    humanize_seconds(self.max_interval),
-                    self.schedule_filename))
+        self.logger.info("Celerybeat: Starting...")
+        self.logger.debug("Celerybeat: "
+            "Ticking with max interval->%s" % (
+                    humanize_seconds(self.max_interval)))
 
         if embedded_process:
             platform.set_process_title("celerybeat")
@@ -199,7 +238,7 @@ class ClockService(object):
                     if self._shutdown.isSet():
                         break
                     interval = self.scheduler.tick()
-                    self.debug("ClockService: Waking up %s." % (
+                    self.debug("Celerybeat: Waking up %s." % (
                             humanize_seconds(interval, prefix="in ")))
                     time.sleep(interval)
             except (KeyboardInterrupt, SystemExit):
@@ -208,32 +247,20 @@ class ClockService(object):
             self.sync()
 
     def sync(self):
-        if self._schedule is not None and not self._in_sync:
-            self.logger.debug("ClockService: Syncing schedule to disk...")
-            self._schedule.sync()
-            self._schedule.close()
-            self._in_sync = True
-            self._stopped.set()
+        self._stopped.set()
 
     def stop(self, wait=False):
-        self.logger.info("ClockService: Shutting down...")
+        self.logger.info("Celerybeat: Shutting down...")
         self._shutdown.set()
         wait and self._stopped.wait() # block until shutdown done.
 
-    @property
-    def schedule(self):
-        if self._schedule is None:
-            filename = self.schedule_filename
-            self._schedule = self.open_schedule(filename=filename)
-        return self._schedule
-
     @property
     def scheduler(self):
         if self._scheduler is None:
-            self._scheduler = self.scheduler_cls(schedule=self.schedule,
-                                            registry=self.registry,
-                                            logger=self.logger,
-                                            max_interval=self.max_interval)
+            self._scheduler = instantiate(self.scheduler_cls,
+                                          schedule=self.schedule,
+                                          logger=self.logger,
+                                          max_interval=self.max_interval)
         return self._scheduler
 
 

celery/bin/celerybeat.py

@@ -8,6 +8,10 @@
     Path to the schedule database. Defaults to ``celerybeat-schedule``.
     The extension ".db" will be appended to the filename.
 
+.. cmdoption:: -S, --scheduler
+
+    Scheduler class to use. Default is celery.beat.Scheduler
+
 .. cmdoption:: -f, --logfile
 
     Path to log file. If no logfile is specified, ``stderr`` is used.
@@ -43,6 +47,13 @@ OPTION_LIST = (
             help="Path to the schedule database. The extension \
                     '.db' will be appended to the filename. Default: %s" % (
                     conf.CELERYBEAT_SCHEDULE_FILENAME)),
+    optparse.make_option('--max-interval',
+            default=3600, type="int", dest="max_interval",
+            help="Maximum time to sleep between rechecking the schedule."),
+    optparse.make_option('-S', '--scheduler',
+            default=None,
+            action="store", dest="scheduler_cls",
+            help="Scheduler class. Default is celery.beat.Scheduler"),
     optparse.make_option('-f', '--logfile', default=conf.CELERYBEAT_LOG_FILE,
             action="store", dest="logfile",
             help="Path to log file."),
@@ -58,13 +69,17 @@ class Beat(object):
 
     def __init__(self, loglevel=conf.CELERYBEAT_LOG_LEVEL,
             logfile=conf.CELERYBEAT_LOG_FILE,
-            schedule=conf.CELERYBEAT_SCHEDULE_FILENAME, **kwargs):
+            schedule=conf.CELERYBEAT_SCHEDULE_FILENAME,
+            max_interval=None,
+            scheduler_cls=None, **kwargs):
         """Starts the celerybeat task scheduler."""
 
         self.loglevel = loglevel
         self.logfile = logfile
         self.schedule = schedule
-        # Setup logging
+        self.scheduler_cls = scheduler_cls
+        self.max_interval = max_interval
+
         if not isinstance(self.loglevel, int):
             self.loglevel = conf.LOG_LEVELS[self.loglevel.upper()]
 
@@ -79,7 +94,9 @@ class Beat(object):
     def start_scheduler(self):
         from celery.log import setup_logger
         logger = setup_logger(self.loglevel, self.logfile, name="celery.beat")
-        beat = self.ClockService(logger,
+        beat = self.ClockService(logger=logger,
+                                 max_interval=self.max_interval,
+                                 scheduler_cls=self.scheduler_cls,
                                  schedule_filename=self.schedule)
 
         try:

celery/conf.py

@@ -62,6 +62,7 @@ _DEFAULTS = {
     "CELERYD_LOG_COLOR": False,
     "CELERYD_LOG_LEVEL": "WARN",
     "CELERYD_LOG_FILE": None, # stderr
+    "CELERYBEAT_SCHEDULE": {},
     "CELERYD_STATE_DB": None,
     "CELERYD_ETA_SCHEDULER_PRECISION": 1,
     "CELERYBEAT_SCHEDULE_FILENAME": "celerybeat-schedule",
@@ -239,6 +240,7 @@ RESULT_PERSISTENT = _get("CELERY_RESULT_PERSISTENT")
 # :--- Celery Beat                                  <-   --   --- - ----- -- #
 CELERYBEAT_LOG_LEVEL = _get("CELERYBEAT_LOG_LEVEL")
 CELERYBEAT_LOG_FILE = _get("CELERYBEAT_LOG_FILE")
+CELERYBEAT_SCHEDULE = _get("CELERYBEAT_SCHEDULE")
 CELERYBEAT_SCHEDULE_FILENAME = _get("CELERYBEAT_SCHEDULE_FILENAME")
 CELERYBEAT_MAX_LOOP_INTERVAL = _get("CELERYBEAT_MAX_LOOP_INTERVAL")
 

celery/schedules.py

@@ -0,0 +1,226 @@
+from datetime import datetime
+from pyparsing import (Word, Literal, ZeroOrMore, Optional,
+                       Group, StringEnd, alphas)
+
+from celery.utils import is_iterable
+from celery.utils.timeutils import timedelta_seconds, weekday, remaining
+
+
+class schedule(object):
+    relative = False
+
+    def __init__(self, run_every=None, relative=False):
+        self.run_every = run_every
+        self.relative = relative
+
+    def remaining_estimate(self, last_run_at):
+        """Returns when the periodic task should run next as a timedelta."""
+        return remaining(last_run_at, self.run_every, relative=self.relative)
+
+    def is_due(self, last_run_at):
+        """Returns tuple of two items ``(is_due, next_time_to_run)``,
+        where next time to run is in seconds.
+
+        See :meth:`celery.task.base.PeriodicTask.is_due` for more information.
+
+        """
+        rem_delta = self.remaining_estimate(last_run_at)
+        rem = timedelta_seconds(rem_delta)
+        if rem == 0:
+            return True, timedelta_seconds(self.run_every)
+        return False, rem
+
+
+class crontab_parser(object):
+    """Parser for crontab expressions. Any expression of the form 'groups' (see
+    BNF grammar below) is accepted and expanded to a set of numbers.  These
+    numbers represent the units of time that the crontab needs to run on::
+
+        digit   :: '0'..'9'
+        dow     :: 'a'..'z'
+        number  :: digit+ | dow+
+        steps   :: number
+        range   :: number ( '-' number ) ?
+        numspec :: '*' | range
+        expr    :: numspec ( '/' steps ) ?
+        groups  :: expr ( ',' expr ) *
+
+    The parser is a general purpose one, useful for parsing hours, minutes and
+    day_of_week expressions.  Example usage::
+
+        minutes = crontab_parser(60).parse("*/15")  # yields [0,15,30,45]
+        hours = crontab_parser(24).parse("*/4")     # yields [0,4,8,12,16,20]
+        day_of_week = crontab_parser(7).parse("*")  # yields [0,1,2,3,4,5,6]
+
+    """
+
+    def __init__(self, max_=60):
+        # define the grammar structure
+        digits = "0123456789"
+        star = Literal('*')
+        number = Word(digits) | Word(alphas)
+        steps = number
+        range_ = number + Optional(Literal('-') + number)
+        numspec = star | range_
+        expr = Group(numspec) + Optional(Literal('/') + steps)
+        extra_groups = ZeroOrMore(Literal(',') + expr)
+        groups = expr + extra_groups + StringEnd()
+
+        # define parse actions
+        star.setParseAction(self._expand_star)
+        number.setParseAction(self._expand_number)
+        range_.setParseAction(self._expand_range)
+        expr.setParseAction(self._filter_steps)
+        extra_groups.setParseAction(self._ignore_comma)
+        groups.setParseAction(self._join_to_set)
+
+        self.max_ = max_
+        self.parser = groups
+
+    @staticmethod
+    def _expand_number(toks):
+        try:
+            i = int(toks[0])
+        except ValueError:
+            try:
+                i = weekday(toks[0])
+            except KeyError:
+                raise ValueError("Invalid weekday literal '%s'." % toks[0])
+        return [i]
+
+    @staticmethod
+    def _expand_range(toks):
+        if len(toks) > 1:
+            return range(toks[0], int(toks[2]) + 1)
+        else:
+            return toks[0]
+
+    def _expand_star(self, toks):
+        return range(self.max_)
+
+    @staticmethod
+    def _filter_steps(toks):
+        numbers = toks[0]
+        if len(toks) > 1:
+            steps = toks[2]
+            return [n for n in numbers if n % steps == 0]
+        else:
+            return numbers
+
+    @staticmethod
+    def _ignore_comma(toks):
+        return filter(lambda x: x != ',', toks)
+
+    @staticmethod
+    def _join_to_set(toks):
+        return set(toks.asList())
+
+    def parse(self, cronspec):
+        return self.parser.parseString(cronspec).pop()
+
+
+class crontab(schedule):
+    """A crontab can be used as the ``run_every`` value of a
+    :class:`PeriodicTask` to add cron-like scheduling.
+
+    Like a :manpage:`cron` job, you can specify units of time of when
+    you would like the task to execute. It is a reasonably complete
+    implementation of cron's features, so it should provide a fair
+    degree of scheduling needs.
+
+    You can specify a minute, an hour, and/or a day of the week in any
+    of the following formats:
+
+    .. attribute:: minute
+
+        - A (list of) integers from 0-59 that represent the minutes of
+          an hour of when execution should occur; or
+        - A string representing a crontab pattern.  This may get pretty
+          advanced, like `minute="*/15"` (for every quarter) or
+          `minute="1,13,30-45,50-59/2"`.
+
+    .. attribute:: hour
+
+        - A (list of) integers from 0-23 that represent the hours of
+          a day of when execution should occur; or
+        - A string representing a crontab pattern.  This may get pretty
+          advanced, like `hour="*/3"` (for every three hours) or
+          `hour="0,8-17/2"` (at midnight, and every two hours during
+          office hours).
+
+    .. attribute:: day_of_week
+
+        - A (list of) integers from 0-6, where Sunday = 0 and Saturday =
+          6, that represent the days of a week that execution should
+          occur.
+        - A string representing a crontab pattern.  This may get pretty
+          advanced, like `day_of_week="mon-fri"` (for weekdays only).
+          (Beware that `day_of_week="*/2"` does not literally mean
+          "every two days", but "every day that is divisible by two"!)
+
+    """
+
+    @staticmethod
+    def _expand_cronspec(cronspec, max_):
+        """Takes the given cronspec argument in one of the forms::
+
+            int         (like 7)
+            basestring  (like '3-5,*/15', '*', or 'monday')
+            set         (like set([0,15,30,45]))
+            list        (like [8-17])
+
+        And convert it to an (expanded) set representing all time unit
+        values on which the crontab triggers.  Only in case of the base
+        type being 'basestring', parsing occurs.  (It is fast and
+        happens only once for each crontab instance, so there is no
+        significant performance overhead involved.)
+
+        For the other base types, merely Python type conversions happen.
+
+        The argument `max_` is needed to determine the expansion of '*'.
+
+        """
+        if isinstance(cronspec, int):
+            result = set([cronspec])
+        elif isinstance(cronspec, basestring):
+            result = crontab_parser(max_).parse(cronspec)
+        elif isinstance(cronspec, set):
+            result = cronspec
+        elif is_iterable(cronspec):
+            result = set(cronspec)
+        else:
+            raise TypeError(
+                    "Argument cronspec needs to be of any of the "
+                    "following types: int, basestring, or an iterable type. "
+                    "'%s' was given." % type(cronspec))
+
+        # assure the result does not exceed the max
+        for number in result:
+            if number >= max_:
+                raise ValueError(
+                        "Invalid crontab pattern. Valid "
+                        "range is 0-%d. '%d' was found." % (max_, number))
+
+        return result
+
+    def __init__(self, minute='*', hour='*', day_of_week='*',
+            nowfun=datetime.now):
+        self.hour = self._expand_cronspec(hour, 24)
+        self.minute = self._expand_cronspec(minute, 60)
+        self.day_of_week = self._expand_cronspec(day_of_week, 7)
+        self.nowfun = nowfun
+
+    def remaining_estimate(self, last_run_at):
+        # remaining_estimate controls the frequency of scheduler
+        # ticks. The scheduler needs to wake up every second in this case.
+        return 1
+
+    def is_due(self, last_run_at):
+        now = self.nowfun()
+        last = now - last_run_at
+        due, when = False, 1
+        if last.days > 0 or last.seconds > 60:
+            due = (now.isoweekday() % 7 in self.day_of_week and
+                   now.hour in self.hour and
+                   now.minute in self.minute)
+        return due, when

celery/task/base.py

@@ -1,20 +1,35 @@
 import sys
+import warnings
+
 from datetime import timedelta
 
 from celery import conf
-from celery.log import setup_task_logger
-from celery.utils.timeutils import timedelta_seconds
-from celery.result import BaseAsyncResult, EagerResult
-from celery.execute import apply_async, apply
-from celery.registry import tasks
 from celery.backends import default_backend
+from celery.exceptions import MaxRetriesExceededError, RetryTaskError
+from celery.execute import apply_async, apply
+from celery.log import setup_task_logger
 from celery.messaging import TaskPublisher, TaskConsumer
 from celery.messaging import establish_connection as _establish_connection
-from celery.exceptions import MaxRetriesExceededError, RetryTaskError
+from celery.registry import tasks
+from celery.result import BaseAsyncResult, EagerResult
+from celery.schedules import schedule
+from celery.utils.timeutils import timedelta_seconds
 
-from celery.task.schedules import schedule
 from celery.task.sets import TaskSet, subtask
 
+PERIODIC_DEPRECATION_TEXT = """\
+Periodic task classes has been deprecated and will be removed
+in celery v3.0.
+
+Please use the CELERYBEAT_SCHEDULE setting instead:
+
+    CELERYBEAT_SCHEDULE = {
+        name: dict(name=task_name, schedule=run_every,
+                   args=(), kwargs={}, options={}, relative=False)
+    }
+
+"""
+
 
 def _unpickle_task(name):
     return tasks[name]
@@ -602,16 +617,16 @@ class PeriodicTask(Task):
             raise NotImplementedError(
                     "Periodic tasks must have a run_every attribute")
 
-        # If run_every is a integer, convert it to timedelta seconds.
-        # Operate on the original class attribute so anyone accessing
-        # it directly gets the right value.
-        if isinstance(self.__class__.run_every, int):
-            self.__class__.run_every = timedelta(seconds=self.run_every)
-
-        # Convert timedelta to instance of schedule.
-        if isinstance(self.__class__.run_every, timedelta):
-            self.__class__.run_every = schedule(self.__class__.run_every,
-                                                self.relative)
+        warnings.warn(PERIODIC_DEPRECATION_TEXT,
+                        PendingDeprecationWarning)
+        conf.CELERYBEAT_SCHEDULE[self.name] = {
+                "name": self.name,
+                "schedule": self.run_every,
+                "args": (),
+                "kwargs": {},
+                "options": {},
+                "relative": self.relative,
+        }
 
         super(PeriodicTask, self).__init__()
 

celery/task/schedules.py

@@ -1,226 +1 @@
-from datetime import datetime
-from pyparsing import (Word, Literal, ZeroOrMore, Optional,
-                       Group, StringEnd, alphas)
-
-from celery.utils import is_iterable
-from celery.utils.timeutils import timedelta_seconds, weekday, remaining
-
-
-class schedule(object):
-    relative = False
-
-    def __init__(self, run_every=None, relative=False):
-        self.run_every = run_every
-        self.relative = relative
-
-    def remaining_estimate(self, last_run_at):
-        """Returns when the periodic task should run next as a timedelta."""
-        return remaining(last_run_at, self.run_every, relative=self.relative)
-
-    def is_due(self, last_run_at):
-        """Returns tuple of two items ``(is_due, next_time_to_run)``,
-        where next time to run is in seconds.
-
-        See :meth:`celery.task.base.PeriodicTask.is_due` for more information.
-
-        """
-        rem_delta = self.remaining_estimate(last_run_at)
-        rem = timedelta_seconds(rem_delta)
-        if rem == 0:
-            return True, timedelta_seconds(self.run_every)
-        return False, rem
-
-
-class crontab_parser(object):
-    """Parser for crontab expressions. Any expression of the form 'groups' (see
-    BNF grammar below) is accepted and expanded to a set of numbers.  These
-    numbers represent the units of time that the crontab needs to run on::
-
-        digit   :: '0'..'9'
-        dow     :: 'a'..'z'
-        number  :: digit+ | dow+
-        steps   :: number
-        range   :: number ( '-' number ) ?
-        numspec :: '*' | range
-        expr    :: numspec ( '/' steps ) ?
-        groups  :: expr ( ',' expr ) *
-
-    The parser is a general purpose one, useful for parsing hours, minutes and
-    day_of_week expressions.  Example usage::
-
-        minutes = crontab_parser(60).parse("*/15")  # yields [0,15,30,45]
-        hours = crontab_parser(24).parse("*/4")     # yields [0,4,8,12,16,20]
-        day_of_week = crontab_parser(7).parse("*")  # yields [0,1,2,3,4,5,6]
-
-    """
-
-    def __init__(self, max_=60):
-        # define the grammar structure
-        digits = "0123456789"
-        star = Literal('*')
-        number = Word(digits) | Word(alphas)
-        steps = number
-        range_ = number + Optional(Literal('-') + number)
-        numspec = star | range_
-        expr = Group(numspec) + Optional(Literal('/') + steps)
-        extra_groups = ZeroOrMore(Literal(',') + expr)
-        groups = expr + extra_groups + StringEnd()
-
-        # define parse actions
-        star.setParseAction(self._expand_star)
-        number.setParseAction(self._expand_number)
-        range_.setParseAction(self._expand_range)
-        expr.setParseAction(self._filter_steps)
-        extra_groups.setParseAction(self._ignore_comma)
-        groups.setParseAction(self._join_to_set)
-
-        self.max_ = max_
-        self.parser = groups
-
-    @staticmethod
-    def _expand_number(toks):
-        try:
-            i = int(toks[0])
-        except ValueError:
-            try:
-                i = weekday(toks[0])
-            except KeyError:
-                raise ValueError("Invalid weekday literal '%s'." % toks[0])
-        return [i]
-
-    @staticmethod
-    def _expand_range(toks):
-        if len(toks) > 1:
-            return range(toks[0], int(toks[2]) + 1)
-        else:
-            return toks[0]
-
-    def _expand_star(self, toks):
-        return range(self.max_)
-
-    @staticmethod
-    def _filter_steps(toks):
-        numbers = toks[0]
-        if len(toks) > 1:
-            steps = toks[2]
-            return [n for n in numbers if n % steps == 0]
-        else:
-            return numbers
-
-    @staticmethod
-    def _ignore_comma(toks):
-        return filter(lambda x: x != ',', toks)
-
-    @staticmethod
-    def _join_to_set(toks):
-        return set(toks.asList())
-
-    def parse(self, cronspec):
-        return self.parser.parseString(cronspec).pop()
-
-
-class crontab(schedule):
-    """A crontab can be used as the ``run_every`` value of a
-    :class:`PeriodicTask` to add cron-like scheduling.
-
-    Like a :manpage:`cron` job, you can specify units of time of when
-    you would like the task to execute. It is a reasonably complete
-    implementation of cron's features, so it should provide a fair
-    degree of scheduling needs.
-
-    You can specify a minute, an hour, and/or a day of the week in any
-    of the following formats:
-
-    .. attribute:: minute
-
-        - A (list of) integers from 0-59 that represent the minutes of
-          an hour of when execution should occur; or
-        - A string representing a crontab pattern.  This may get pretty
-          advanced, like `minute="*/15"` (for every quarter) or
-          `minute="1,13,30-45,50-59/2"`.
-
-    .. attribute:: hour
-
-        - A (list of) integers from 0-23 that represent the hours of
-          a day of when execution should occur; or
-        - A string representing a crontab pattern.  This may get pretty
-          advanced, like `hour="*/3"` (for every three hours) or
-          `hour="0,8-17/2"` (at midnight, and every two hours during
-          office hours).
-
-    .. attribute:: day_of_week
-
-        - A (list of) integers from 0-6, where Sunday = 0 and Saturday =
-          6, that represent the days of a week that execution should
-          occur.
-        - A string representing a crontab pattern.  This may get pretty
-          advanced, like `day_of_week="mon-fri"` (for weekdays only).
-          (Beware that `day_of_week="*/2"` does not literally mean
-          "every two days", but "every day that is divisible by two"!)
-
-    """
-
-    @staticmethod
-    def _expand_cronspec(cronspec, max_):
-        """Takes the given cronspec argument in one of the forms::
-
-            int         (like 7)
-            basestring  (like '3-5,*/15', '*', or 'monday')
-            set         (like set([0,15,30,45]))
-            list        (like [8-17])
-
-        And convert it to an (expanded) set representing all time unit
-        values on which the crontab triggers.  Only in case of the base
-        type being 'basestring', parsing occurs.  (It is fast and
-        happens only once for each crontab instance, so there is no
-        significant performance overhead involved.)
-
-        For the other base types, merely Python type conversions happen.
-
-        The argument `max_` is needed to determine the expansion of '*'.
-
-        """
-        if isinstance(cronspec, int):
-            result = set([cronspec])
-        elif isinstance(cronspec, basestring):
-            result = crontab_parser(max_).parse(cronspec)
-        elif isinstance(cronspec, set):
-            result = cronspec
-        elif is_iterable(cronspec):
-            result = set(cronspec)
-        else:
-            raise TypeError(
-                    "Argument cronspec needs to be of any of the "
-                    "following types: int, basestring, or an iterable type. "
-                    "'%s' was given." % type(cronspec))
-
-        # assure the result does not exceed the max
-        for number in result:
-            if number >= max_:
-                raise ValueError(
-                        "Invalid crontab pattern. Valid "
-                        "range is 0-%d. '%d' was found." % (max_, number))
-
-        return result
-
-    def __init__(self, minute='*', hour='*', day_of_week='*',
-            nowfun=datetime.now):
-        self.hour = self._expand_cronspec(hour, 24)
-        self.minute = self._expand_cronspec(minute, 60)
-        self.day_of_week = self._expand_cronspec(day_of_week, 7)
-        self.nowfun = nowfun
-
-    def remaining_estimate(self, last_run_at):
-        # remaining_estimate controls the frequency of scheduler
-        # ticks. The scheduler needs to wake up every second in this case.
-        return 1
-
-    def is_due(self, last_run_at):
-        now = self.nowfun()
-        last = now - last_run_at
-        due, when = False, 1
-        if last.days > 0 or last.seconds > 60:
-            due = (now.isoweekday() % 7 in self.day_of_week and
-                   now.hour in self.hour and
-                   now.minute in self.minute)
-        return due, when
+from celery.schedules import schedule, crontab_parser, crontab