Ticket #10154: 10154-r14446.diff

django/db/models/expressions.py

@@ -1 @@
-from datetime import datetime
+import datetime
 
 from django.utils import tree
 from django.utils.copycompat import deepcopy
@@ -26 +26 @@
         super(ExpressionNode, self).__init__(children, connector, negated)
 
     def _combine(self, other, connector, reversed, node=None):
+        if isinstance(other, datetime.timedelta):
+            return DateModifierNode([self, other], connector)
+
         if reversed:
             obj = ExpressionNode([other], connector)
             obj.add(node or self, connector)
@@ -111 +114 @@
 
     def evaluate(self, evaluator, qn, connection):
         return evaluator.evaluate_leaf(self, qn, connection)
+
+class DateModifierNode(ExpressionNode):
+    """
+    Node that implements the following syntax:
+    filter(end_date__gt=F('start_date') + datetime.timedelta(days=3, seconds=200))
+
+    which translates into:
+    POSTGRES:
+        WHERE end_date > (start_date + INTERVAL '3 days 200 seconds')
+
+    MYSQL:
+        WHERE end_date > (start_date + INTERVAL '3 0:0:200:0' DAY_MICROSECOND)
+
+    ORACLE:
+        WHERE end_date > (start_date + INTERVAL '3 00:03:20.000000' DAY(1) TO SECOND(6))
+
+    SQLITE:
+        WHERE end_date > django_format_dtdelta(start_date, "+" "3", "200", "0")
+        (A custom function is used in order to preserve six digits of fractional
+        second information on sqlite, and to format both date and datetime values.)
+
+    Note that microsecond comparisons are not well supported with MySQL, since 
+    MySQL does not store microsecond information.
+
+    Only adding and subtracting timedeltas is supported, attempts to use other 
+    operations raise a TypeError.
+    """
+    def __init__(self, children, connector, negated=False):
+        if len(children) != 2:
+            raise TypeError('Must specify a node and a timedelta.')
+        if not isinstance(children[1], datetime.timedelta):
+            raise TypeError('Second child must be a timedelta.')
+        if connector not in (self.ADD, self.SUB):
+            raise TypeError('Connector must be + or -, not %s' % connector)
+        super(DateModifierNode, self).__init__(children, connector, negated)
+
+    def evaluate(self, evaluator, qn, connection):
+        timedelta = self.children.pop()
+        sql, params = evaluator.evaluate_node(self, qn, connection)
+
+        if timedelta.days == 0 and timedelta.seconds == 0 and \
+                timedelta.microseconds == 0:
+            return sql, params
+
+        return connection.ops.date_interval_sql(sql, self.connector, timedelta), params

django/db/backends/postgresql/operations.py

@@ -27 +27 @@
         else:
             return "EXTRACT('%s' FROM %s)" % (lookup_type, field_name)
 
+    def date_interval_sql(self, sql, connector, timedelta):
+        """
+        implements the interval functionality for expressions
+        format for Postgres:
+            (datefield + interval '3 days 200 seconds 5 microseconds')
+        """
+        modifiers = []
+        if timedelta.days:
+            modifiers.append(u'%s days' % timedelta.days)
+        if timedelta.seconds:
+            modifiers.append(u'%s seconds' % timedelta.seconds)
+        if timedelta.microseconds:
+            modifiers.append(u'%s microseconds' % timedelta.microseconds)
+        mods = u' '.join(modifiers)
+        conn = u' %s ' % connector
+        return u'(%s)' % conn.join([sql, u'interval \'%s\'' % mods])
+
     def date_trunc_sql(self, lookup_type, field_name):
         # http://www.postgresql.org/docs/8.0/static/functions-datetime.html#FUNCTIONS-DATETIME-TRUNC
         return "DATE_TRUNC('%s', %s)" % (lookup_type, field_name)

django/db/backends/sqlite3/base.py

@@ -9 +9 @@
 
 import re
 import sys
+import datetime
 
 from django.db import utils
 from django.db.backends import *
@@ -90 +91 @@
         # cause a collision with a field name).
         return "django_extract('%s', %s)" % (lookup_type.lower(), field_name)
 
+    def date_interval_sql(self, sql, connector, timedelta):
+        # It would be more straightforward if we could use the sqlite strftime
+        # function, but it does not allow for keeping six digits of fractional
+        # second information, nor does it allow for formatting date and datetime
+        # values differently. So instead we register our own function that 
+        # formats the datetime combined with the delta in a manner suitable 
+        # for comparisons.
+        return  u'django_format_dtdelta(%s, "%s", "%d", "%d", "%d")' % (sql, 
+            connector, timedelta.days, timedelta.seconds, timedelta.microseconds)
+
     def date_trunc_sql(self, lookup_type, field_name):
         # sqlite doesn't support DATE_TRUNC, so we fake it with a user-defined
         # function django_date_trunc that's registered in connect(). Note that
@@ -197 +208 @@
             self.connection.create_function("django_extract", 2, _sqlite_extract)
             self.connection.create_function("django_date_trunc", 2, _sqlite_date_trunc)
             self.connection.create_function("regexp", 2, _sqlite_regexp)
+            self.connection.create_function("django_format_dtdelta", 5, _sqlite_format_dtdelta)
             connection_created.send(sender=self.__class__, connection=self)
         return self.connection.cursor(factory=SQLiteCursorWrapper)
 
@@ -260 +272 @@
     elif lookup_type == 'day':
         return "%i-%02i-%02i 00:00:00" % (dt.year, dt.month, dt.day)
 
+def _sqlite_format_dtdelta(dt, conn, days, secs, usecs):
+    try:
+        dt = util.typecast_timestamp(dt)
+        delta = datetime.timedelta(int(days), int(secs), int(usecs))
+        if conn.strip() == '+':
+            dt = dt + delta
+        else:
+            dt = dt - delta
+    except (ValueError, TypeError):
+        return None
+
+    if isinstance(dt, datetime.datetime):
+        rv = dt.strftime("%Y-%m-%d %H:%M:%S")
+        if dt.microsecond:
+            rv = "%s.%0.6d" % (rv, dt.microsecond)
+    else:
+        rv = dt.strftime("%Y-%m-%d")
+    return rv
+
 def _sqlite_regexp(re_pattern, re_string):
     import re
     try:

django/db/backends/mysql/base.py

@@ -158 +158 @@
             sql = "CAST(DATE_FORMAT(%s, '%s') AS DATETIME)" % (field_name, format_str)
         return sql
 
+    def date_interval_sql(self, sql, connector, timedelta):
+        return "(%s %s INTERVAL '%d 0:0:%d:%d' DAY_MICROSECOND)" % (sql, connector, 
+                timedelta.days, timedelta.seconds, timedelta.microseconds)
+
     def drop_foreignkey_sql(self):
         return "DROP FOREIGN KEY"
 

django/db/backends/oracle/base.py

@@ -95 +95 @@
         else:
             return "EXTRACT(%s FROM %s)" % (lookup_type, field_name)
 
+    def date_interval_sql(self, sql, connector, timedelta):
+        """
+        Implements the interval functionality for expressions
+        format for Oracle:
+        (datefield + INTERVAL '3 00:03:20.000000' DAY(1) TO SECOND(6))
+        """
+        minutes, seconds = divmod(timedelta.seconds, 60)
+        hours, minutes = divmod(minutes, 60)
+        days = str(timedelta.days)
+        day_precision = len(days)
+        fmt = "(%s %s INTERVAL '%s %02d:%02d:%02d.%06d' DAY(%d) TO SECOND(6))"
+        return fmt % (sql, connector, days, hours, minutes, seconds, 
+                timedelta.microseconds, day_precision)
+
     def date_trunc_sql(self, lookup_type, field_name):
         # Oracle uses TRUNC() for both dates and numbers.
         # http://download-east.oracle.com/docs/cd/B10501_01/server.920/a96540/functions155a.htm#SQLRF06151

django/db/backends/__init__.py

@@ -216 +216 @@
         """
         raise NotImplementedError()
 
+    def date_interval_sql(self, sql, connector, timedelta):
+        """
+        Implements the date interval functionality for expressions
+        """
+        raise NotImplementedError()
+
     def date_trunc_sql(self, lookup_type, field_name):
         """
         Given a lookup_type of 'year', 'month' or 'day', returns the SQL that

tests/regressiontests/expressions_regress/tests.py

@@ +1 @@
+import datetime
+
+from django.test import TestCase
+from django.db.models import F
+from django.conf import settings
+from django.db import DEFAULT_DB_ALIAS
+
+from models import Experiment
+
+class FTimeDeltaTests(TestCase):
+
+    def setUp(self):
+        db = settings.DATABASES[DEFAULT_DB_ALIAS]
+        self.db_is_mysql = db['ENGINE'] == 'django.db.backends.mysql'
+        self.db_is_sqlite = db['ENGINE'] == 'django.db.backends.sqlite3'
+
+        sday = datetime.date(2010, 6, 25)
+        stime = datetime.datetime(2010, 6, 25, 12, 15, 30, 747000)
+        midnight = datetime.time(0)
+
+        delta0 = datetime.timedelta(0)
+        delta1 = datetime.timedelta(microseconds=253000)
+        delta2 = datetime.timedelta(seconds=44)
+        delta3 = datetime.timedelta(hours=21, minutes=8)
+        delta4 = datetime.timedelta(days=10)
+
+        # Test data is set so that deltas and delays will be 
+        # strictly increasing.
+        self.deltas = []
+        self.delays = []
+        self.days_long = []
+
+        # e0: started same day as assigned, zero duration
+        end = stime+delta0
+        e0 = Experiment.objects.create(name='e0', assigned=sday, start=stime, 
+            end=end, completed=end.date())
+        self.deltas.append(delta0)
+        self.delays.append(e0.start-
+            datetime.datetime.combine(e0.assigned, midnight))
+        self.days_long.append(e0.completed-e0.assigned)
+
+        # e1: started one day after assigned, tiny duration, data
+        # set so that end time has no fractional seconds, which
+        # tests an edge case on sqlite. This Experiment is only
+        # included in the test data when DB is not MySQL (since
+        # on MySQL microseconds are dropped from datetime fields).
+        if not self.db_is_mysql:
+            delay = datetime.timedelta(1)
+            end = stime + delay + delta1
+            e1 = Experiment.objects.create(name='e1', assigned=sday, 
+                start=stime+delay, end=end, completed=end.date())
+            self.deltas.append(delta1)
+            self.delays.append(e1.start-
+                datetime.datetime.combine(e1.assigned, midnight))
+            self.days_long.append(e1.completed-e1.assigned)
+
+        # e2: started three days after assigned, small duration
+        end = stime+delta2
+        e2 = Experiment.objects.create(name='e2',
+            assigned=sday-datetime.timedelta(3), start=stime, end=end, 
+            completed=end.date())
+        self.deltas.append(delta2)
+        self.delays.append(e2.start-
+            datetime.datetime.combine(e2.assigned, midnight))
+        self.days_long.append(e2.completed-e2.assigned)
+
+        # e3: started four days after assigned, medium duration
+        delay = datetime.timedelta(4)
+        end = stime + delay + delta3
+        e3 = Experiment.objects.create(name='e3',
+            assigned=sday, start=stime+delay, end=end, completed=end.date())
+        self.deltas.append(delta3)
+        self.delays.append(e3.start-
+            datetime.datetime.combine(e3.assigned, midnight))
+        self.days_long.append(e3.completed-e3.assigned)
+
+        # e4: started 10 days after assignment, long duration
+        end = stime + delta4
+        e4 = Experiment.objects.create(name='e4',
+            assigned=sday-datetime.timedelta(10), start=stime, end=end,
+            completed=end.date())
+        self.deltas.append(delta4)
+        self.delays.append(e4.start-
+            datetime.datetime.combine(e4.assigned, midnight))
+        self.days_long.append(e4.completed-e4.assigned)
+        self.expnames = [e.name for e in Experiment.objects.all()]
+
+    def test_delta_add(self):
+        for i in range(len(self.deltas)):
+            delta = self.deltas[i]
+            test_set = [e.name for e in
+                Experiment.objects.filter(end__lt=F('start')+delta)]
+            self.assertEqual(test_set, self.expnames[:i])
+
+            test_set = [e.name for e in
+                Experiment.objects.filter(end__lte=F('start')+delta)]
+            self.assertEqual(test_set, self.expnames[:i+1])
+
+    def test_delta_subtract(self):
+        for i in range(len(self.deltas)):
+            delta = self.deltas[i]
+            test_set = [e.name for e in
+                Experiment.objects.filter(start__gt=F('end')-delta)]
+            self.assertEqual(test_set, self.expnames[:i])
+
+            test_set = [e.name for e in
+                Experiment.objects.filter(start__gte=F('end')-delta)]
+            self.assertEqual(test_set, self.expnames[:i+1])
+
+    def test_exclude(self):
+        for i in range(len(self.deltas)):
+            delta = self.deltas[i]
+            test_set = [e.name for e in 
+                Experiment.objects.exclude(end__lt=F('start')+delta)]
+            self.assertEqual(test_set, self.expnames[i:])
+
+            test_set = [e.name for e in 
+                Experiment.objects.exclude(end__lte=F('start')+delta)]
+            self.assertEqual(test_set, self.expnames[i+1:])
+
+    def test_date_comparison(self):
+        for i in range(len(self.days_long)):
+            days = self.days_long[i]
+            test_set = [e.name for e in 
+                Experiment.objects.filter(completed__lt=F('assigned')+days)]
+            self.assertEqual(test_set, self.expnames[:i])
+
+            test_set = [e.name for e in 
+                Experiment.objects.filter(completed__lte=F('assigned')+days)]
+            self.assertEqual(test_set, self.expnames[:i+1])
+
+    def test_mixed_comparisons1(self):
+        for i in range(len(self.delays)):
+            delay = self.delays[i]
+            if self.db_is_mysql:
+                delay = datetime.timedelta(delay.days, delay.seconds)
+            test_set = [e.name for e in
+                Experiment.objects.filter(assigned__gt=F('start')-delay)]
+            self.assertEqual(test_set, self.expnames[:i])
+
+            test_set = [e.name for e in
+                Experiment.objects.filter(assigned__gte=F('start')-delay)]
+            self.assertEqual(test_set, self.expnames[:i+1])
+
+    def test_mixed_comparisons2(self):
+        delays = [datetime.timedelta(delay.days) for delay in self.delays]
+        for i in range(len(delays)):
+            delay = delays[i]
+            test_set = [e.name for e in
+                Experiment.objects.filter(start__lt=F('assigned')+delay)]
+            self.assertEqual(test_set, self.expnames[:i])
+
+            test_set = [e.name for e in
+                Experiment.objects.filter(start__lte=F('assigned')+delay+
+                    datetime.timedelta(1))]
+            self.assertEqual(test_set, self.expnames[:i+1])
+
+    def test_delta_update(self):
+        for i in range(len(self.deltas)):
+            delta = self.deltas[i]
+            exps = Experiment.objects.all()
+            expected_durations = [e.duration() for e in exps]
+            expected_starts = [e.start+delta for e in exps]
+            expected_ends = [e.end+delta for e in exps]
+
+            Experiment.objects.update(start=F('start')+delta, end=F('end')+delta)
+            exps = Experiment.objects.all()
+            new_starts = [e.start for e in exps]
+            new_ends = [e.end for e in exps]
+            new_durations = [e.duration() for e in exps]
+            self.assertEqual(expected_starts, new_starts)
+            self.assertEqual(expected_ends, new_ends)
+            self.assertEqual(expected_durations, new_durations)
+
+    def test_delta_invalid_op_mult(self):
+        raised = False
+        try:
+            r = repr(Experiment.objects.filter(end__lt=F('start')*self.deltas[0]))
+        except TypeError:
+            raised = True
+        self.assertTrue(raised, "TypeError not raised on attempt to multiply datetime by timedelta.")
+
+    def test_delta_invalid_op_div(self):
+        raised = False
+        try:
+            r = repr(Experiment.objects.filter(end__lt=F('start')/self.deltas[0]))
+        except TypeError:
+            raised = True
+        self.assertTrue(raised, "TypeError not raised on attempt to divide datetime by timedelta.")
+
+    def test_delta_invalid_op_mod(self):
+        raised = False
+        try:
+            r = repr(Experiment.objects.filter(end__lt=F('start')%self.deltas[0]))
+        except TypeError:
+            raised = True
+        self.assertTrue(raised, "TypeError not raised on attempt to modulo divide datetime by timedelta.")
+
+    def test_delta_invalid_op_and(self):
+        raised = False
+        try:
+            r = repr(Experiment.objects.filter(end__lt=F('start')&self.deltas[0]))
+        except TypeError:
+            raised = True
+        self.assertTrue(raised, "TypeError not raised on attempt to binary and a datetime with a timedelta.")
+
+    def test_delta_invalid_op_or(self):
+        raised = False
+        try:
+            r = repr(Experiment.objects.filter(end__lt=F('start')|self.deltas[0]))
+        except TypeError:
+            raised = True
+        self.assertTrue(raised, "TypeError not raised on attempt to binary or a datetime with a timedelta.")
+
 """
 Spanning tests for all the operations that F() expressions can perform.
 """

tests/regressiontests/expressions_regress/models.py

@@ -10 +10 @@
     def __unicode__(self):
         return u'%i, %.3f' % (self.integer, self.float)
 
+class Experiment(models.Model): 
+    name = models.CharField(max_length=24) 
+    assigned = models.DateField() 
+    completed = models.DateField()
+    start = models.DateTimeField() 
+    end = models.DateTimeField() 
+     
+    class Meta: 
+        ordering = ('name',) 
+ 
+    def duration(self): 
+        return self.end - self.start 
+

docs/topics/db/queries.txt

@@ -36 +36 @@
         headline = models.CharField(max_length=255)
         body_text = models.TextField()
         pub_date = models.DateTimeField()
+        mod_date = models.DateTimeField()
         authors = models.ManyToManyField(Author)
         n_comments = models.IntegerField()
         n_pingbacks = models.IntegerField()
@@ -538 +539 @@
 
     >>> Entry.objects.filter(authors__name=F('blog__name'))
 
+For date fields, you can add or subtract a ``datetime.timedelta`` object.  The
+following would return all entries that were modified more than 3 days after
+they were published:
+   
+    >>> from datetime import timedelta
+    >>> Entry.objects.filter(mod_date__gt=F('pub_date') + timedelta(days=3))
+
 The pk lookup shortcut
 ----------------------