celerybeat: Added Interface required for the Django database scheduler to work.

Author: Ask Solem

celery/beat.py

@@ -16,6 +16,7 @@
 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
 
 
@@ -84,6 +85,12 @@ def is_due(self):
         """See :meth:`celery.task.base.PeriodicTask.is_due`."""
         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.
@@ -105,6 +112,7 @@ class Scheduler(UserDict):
         Maximum time to sleep between re-checking the schedule.
 
     """
+    Entry = ScheduleEntry
 
     def __init__(self, schedule=None, logger=None,
             max_interval=None):
@@ -117,6 +125,9 @@ def __init__(self, schedule=None, logger=None,
         self.cleanup()
         self.setup_schedule()
 
+    def iterentries(self):
+        return self.schedule.itervalues()
+
     def tick(self):
         """Run a tick, that is one iteration of the scheduler.
         Executes all due tasks."""
@@ -126,7 +137,7 @@ def tick(self):
         remaining_times = []
         connection = establish_connection()
         try:
-            for entry in self.schedule.values():
+            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)
@@ -145,12 +156,17 @@ def tick(self):
 
         return min(remaining_times + [self.max_interval])
 
-    def apply_async(self, entry, **kwargs):
+    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()
+        entry = self.reserve(entry)
+
+        print("APPLYING: %s" % (entry, ))
 
         try:
             result = send_task(entry.name, entry.args, entry.kwargs,
@@ -168,15 +184,15 @@ def maybe_schedule(self, s, relative=False):
         return s
 
     def setup_schedule(self):
-        self.schedule = self.dict_to_entries(conf.CELERYBEAT_SCHEDULE)
+        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] = ScheduleEntry(**entry)
+            entries[name] = self.Entry(**entry)
         return entries
 
     def cleanup(self):
@@ -194,9 +210,11 @@ class ClockService(object):
     def __init__(self, logger=None,
             max_interval=conf.CELERYBEAT_MAX_LOOP_INTERVAL,
             schedule=conf.CELERYBEAT_SCHEDULE,
-            schedule_filename=conf.CELERYBEAT_SCHEDULE_FILENAME):
+            schedule_filename=conf.CELERYBEAT_SCHEDULE_FILENAME,
+            scheduler_cls=None):
         self.logger = logger or log.get_default_logger()
         self.max_interval = max_interval
+        self.scheduler_cls = scheduler_cls or self.scheduler_cls
         self._shutdown = threading.Event()
         self._stopped = threading.Event()
         self.schedule = schedule
@@ -239,9 +257,10 @@ def stop(self, wait=False):
     @property
     def scheduler(self):
         if self._scheduler is None:
-            self._scheduler = self.scheduler_cls(schedule=self.schedule,
-                                            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 @@
             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 @@ def run(self):
     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/task/base.py

@@ -19,7 +19,7 @@
 
 PERIODIC_DEPRECATION_TEXT = """\
 Periodic task classes has been deprecated and will be removed
-in celery v1.6.
+in celery v3.0.
 
 Please use the CELERYBEAT_SCHEDULE setting instead: