Rewrote schedules

Schedules are cleaned up a lot. They no longer know about actual
task instances. Much less repetition and extra arguments allow you
to specify more precise times.

Author: matthewwithanm

periodically/backends.py

@@ -1,6 +1,7 @@
 from .models import ExecutionRecord
 from datetime import datetime
 from .signals import task_complete
+from .utils import get_scheduled_time
 import logging
 import sys
 
@@ -11,6 +12,7 @@
 class BaseBackend(object):
     """
     Keeps a schedule of periodic tasks.
+
     """
     _schedules = []
 
@@ -83,14 +85,12 @@ def _run_tasks(self, tasks=None, fake=None, force=False):
                     return
 
                 # Run the task if it's due (or past due).
-                if force or schedule.get_scheduled_time(task) <= datetime.now():
-                    previous_record = schedule.get_previous_record(task)
+                if force or get_scheduled_time(task, schedule) <= datetime.now():
+                    previous_record = ExecutionRecord.objects.get_most_recent(task, schedule)
                     fake_task = fake or fake is None and previous_record is None
 
                     # If we're forcing the task, use the previous scheduled
-                    # time. That way we don't put off any upcoming tasks whose
-                    # schedule's get_scheduled_time method relies on the time
-                    # of the previous execution.
+                    # time.
                     scheduled_time = previous_record.scheduled_time if force and previous_record else None
 
                     if fake_task:
@@ -100,7 +100,8 @@ def _run_tasks(self, tasks=None, fake=None, force=False):
 
     def check_timeout(self, task):
         from .settings import DEFAULT_TIMEOUT
-        for record in ExecutionRecord.objects.filter(task_id=task.task_id, end_time__isnull=True):
+        for record in ExecutionRecord.objects.filter(task_id=task.task_id,
+                end_time__isnull=True):
             timeout = getattr(task, 'timeout', DEFAULT_TIMEOUT)
             running_time = datetime.now() - record.start_time
             if running_time > timeout:
@@ -124,7 +125,7 @@ def fake_task(self, task, schedule, scheduled_time=None):
         self.logger.info('Faking periodic task "%s"' % task.task_id)
 
         if scheduled_time is None:
-            scheduled_time = schedule.get_scheduled_time(task)
+            scheduled_time = get_scheduled_time(task, schedule)
 
         # Create the log for this execution.
         log = ExecutionRecord.objects.create(
@@ -150,7 +151,7 @@ def run_task(self, task, schedule, scheduled_time=None):
 
 
         if scheduled_time is None:
-            scheduled_time = schedule.get_scheduled_time(task)
+            scheduled_time = get_scheduled_time(task, schedule)
 
         # Create the log for this execution.
         log = ExecutionRecord.objects.create(
@@ -195,11 +196,14 @@ def complete_task(self, task, success=True, extra=None):
         Marks a task as complete and performs other post-completion tasks. The
         <code>extra</code> argument is a dictionary of values to be passed to
         <code>Logger.log()</code> as keyword args.
+
         """
         if extra is not None:
             self.logger.log(**extra)
 
-        record = ExecutionRecord.objects.filter(task_id=task.task_id, end_time=None).order_by('-start_time')[0]
+        record = ExecutionRecord.objects \
+                .filter(task_id=task.task_id, end_time=None) \
+                .order_by('-start_time')[0]
         record.end_time = datetime.now()
         record.completed_successfully = success
         record.save()
@@ -211,5 +215,6 @@ class DefaultBackend(BaseBackend):
     """
     A backend that only runs tasks when explicitly told to (i.e. when its
     `run_scheduled_tasks()` method is invoked).
+
     """
     pass

periodically/models.py

@@ -1,6 +1,16 @@
 from django.db import models
 
 
+class ExecutionRecordManager(models.Manager):
+    def get_most_recent(self, task=None, schedule=None):
+        qs = self.get_query_set().order_by('-start_time').all()
+        if task:
+            qs = qs.filter(task_id=task.task_id)
+        if schedule:
+            qs = qs.filter(schedule_id=schedule.schedule_id)
+        return qs[0] if qs else None
+
+
 class ExecutionRecord(models.Model):
     task_id = models.CharField(max_length=255)
     schedule_id = models.BigIntegerField()
@@ -9,6 +19,8 @@ class ExecutionRecord(models.Model):
     end_time = models.DateTimeField(blank=True, null=True)
     completed_successfully = models.BooleanField(default=False)
     is_fake = models.BooleanField(default=False)
-    
+
+    objects = ExecutionRecordManager()
+
     def __unicode__(self):
         return '%s @ %s' % (self.task_id, self.start_time)

periodically/schedules.py

@@ -1,106 +1,115 @@
+import time as _time
 from datetime import datetime, date, time, timedelta
-from .models import ExecutionRecord
 
 
 class Schedule(object):
-
-    def get_previous_record(self, task):
-        record_list = ExecutionRecord.objects \
-                .filter(task_id=task.task_id, schedule_id=self.__hash__()) \
-                .order_by('-start_time')
-        if record_list:
-            return record_list[0]
-        else:
-            return None
-
     def __eq__(self, other):
-        return self.__dict__ == other.__dict__
+        other_id = getattr(other, 'schedule_id', None)
+        return self.schedule_id == other_id
 
     def __ne__(self, other):
         return not self.__eq__(other)
 
+    def __hash__(self):
+        return self.schedule_id
+
+
+class BaseSchedule(object):
+    """A base class for the built-in schedules. Don't use this yourself.
+
+    """
 
-class PeriodicSchedule(Schedule):
+    @property
+    def schedule_id(self):
+        class_name = '%s.%s' % (self.__class__.__module__,
+                self.__class__.__name__)
+        time_args = tuple([getattr(self, name) for name in self._time_attrs])
+        return hash((class_name, time_args))
 
-    def get_scheduled_time(self, task):
-        previous_record = self.get_previous_record(task)
-        if previous_record:
-            next_run_time = previous_record.scheduled_time \
-                    + self.repeat_interval
+    def time_before(self, time):
+        kwargs = dict((k, getattr(self, k)) for k in self._time_attrs)
+        t = time.replace(**kwargs)
+        if t > time:
+            previous_time = t - self.repeat_interval
         else:
-            next_run_time = datetime.now()
-        return next_run_time
+            previous_time = t
+        return previous_time
 
-    def __hash__(self):
-        class_name = '%s.%s' % (self.__module__, self.__class__)
-        return hash((class_name, self.repeat_interval))
+    def time_after(self, time):
+        return self.time_before(time) + self.repeat_interval
 
 
-class Daily(Schedule):
+class Hourly(BaseSchedule):
+    repeat_interval = timedelta(hours=1)
+    _time_attrs = ['minute', 'second', 'microsecond']
 
-    def __init__(self, hour=0, minute=0, second=0, microsecond=0):
-        self.hour = hour
+    def __init__(self, minute=0, second=0, microsecond=0):
+        super(Hourly, self).__init__()
         self.minute = minute
         self.second = second
         self.microsecond = microsecond
 
-    def __hash__(self):
-        class_name = '%s.%s' % (self.__module__, self.__class__)
-        return hash((class_name, self.hour, self.minute, self.second,
-                self.microsecond))
-
-
-    def get_scheduled_time(self, task):
-        previous_record = self.get_previous_record(task)
-        today = date.today()
-        t = time(self.hour, self.minute, self.second, self.microsecond)
-        now = datetime.now()
-
-        if previous_record:
-            previous_run_date = \
-                    datetime.date(previous_record.scheduled_time)
-            scheduled_date = previous_run_date + timedelta(days=1)
-            scheduled_time = datetime.combine(scheduled_date, t)
-            use_last_skipped_time = scheduled_time < now
-        else:
-            use_last_skipped_time = True
-
-        # If the next scheduled time already happened, just use the last
-        # skipped time. Otherwise, we'd end up scheduling an execution for
-        # each missed day.
-        if use_last_skipped_time:
-            todays_run_time = datetime.combine(today, t)
-            if todays_run_time < now:
-                scheduled_time = todays_run_time
-            else:
-                scheduled_time = todays_run_time - timedelta(days=1)
+    def __unicode__(self):
+        return 'Hourly @ %s' % time(minute=self.minute, second=self.second,
+                microsecond=self.microsecond)
 
-        return scheduled_time
 
+class Daily(BaseSchedule):
+    repeat_interval = timedelta(days=1)
+    _time_attrs = ['hour', 'minute', 'second', 'microsecond']
 
-class Hourly(PeriodicSchedule):
-    repeat_interval = timedelta(hours=1)
+    def __init__(self, hour=0, minute=0, second=0, microsecond=0):
+        super(Daily, self).__init__()
+        self.hour = hour
+        self.minute = minute
+        self.second = second
+        self.microsecond = microsecond
 
     def __unicode__(self):
-        return 'Hourly'
+        return 'Daily @ %s' % time(hour=self.hour, minute=self.minute,
+                second=self.second, microsecond=self.microsecond)
+
 
+class Weekly(BaseSchedule):
+    repeat_interval = timedelta(days=7)
+    _time_attrs = ['day', 'hour', 'minute', 'second', 'microsecond']
 
-class Weekly(PeriodicSchedule):
-    repeat_interval = timedelta(weeks=1)
+    def __init__(self, day=0, hour=0, minute=0, second=0, microsecond=0):
+        super(Weekly, self).__init__()
+        self.day = day
+        self.hour = hour
+        self.minute = minute
+        self.second = second
+        self.microsecond = microsecond
 
     def __unicode__(self):
-        return 'Weekly'
+        day_name = _time.strftime('%A', _time.strptime(str(self.day), '%w'))
+        return '%s @ %s' % (day_name, time(hour=self.hour, minute=self.minute,
+                second=self.second, microsecond=self.microsecond))
 
 
-class Every(PeriodicSchedule):
+class Every(BaseSchedule):
+    _time_attrs = ['repeat_interval']
 
     def __init__(self, interval=None, days=0, seconds=0, microseconds=0,
-            milliseconds=0, minutes=0, hours=0, weeks=0):
+            milliseconds=0, minutes=0, hours=0, weeks=0,
+            starting_at=datetime(1970, 1, 1)):
         super(Every, self).__init__()
         if interval is None:
             interval = timedelta(days, seconds, microseconds,
                 milliseconds, minutes, hours, weeks)
         self.repeat_interval = interval
+        self.starting_at = starting_at
+
+    def time_before(self, time):
+        # Uses "perfect time" (no leap years, etc.)
+        interval = self.repeat_interval.microseconds / 1000 \
+                + self.repeat_interval.total_seconds() * 1000
+        time_since_start = time - self.starting_at
+        ms_since_start = time_since_start.microseconds / 1000 \
+                + time_since_start.total_seconds() * 1000
+        time = interval * int(ms_since_start / interval)
+        return datetime.fromtimestamp(time / 1000)
 
     def __unicode__(self):
         return 'Every %s' % self.repeat_interval

periodically/utils.py

@@ -1,5 +1,7 @@
+from datetime import datetime
 from django.utils import importlib
 from . import settings
+from .models import ExecutionRecord
 
 
 class InvalidBackendAliasError(Exception):
@@ -12,6 +14,20 @@ def __init__(self, backend):
         Exception.__init__(self, 'Could not find backend %s' % backend)
 
 
+def get_scheduled_time(task, schedule, now=None):
+    previous_record = ExecutionRecord.objects.get_most_recent(task, schedule)
+    if not now:
+        now = datetime.now()
+    previous_scheduled_time = schedule.time_before(now)
+    if not previous_record:
+        scheduled_time = previous_scheduled_time
+    elif previous_record.scheduled_time < previous_scheduled_time:
+        scheduled_time = previous_scheduled_time
+    else:
+        scheduled_time = schedule.time_after(now)
+    return scheduled_time    
+
+
 def get_scheduler_backend_class(backend_alias=None):
     """
     Accepts a scheduler alias and returns the corresponding backend class.