Add restore_history to ModelSQL

issue3456
review1441003

CHANGELOG

@@ -1,3 +1,4 @@
+* Add restore_history to ModelSQL
 * Add history revisions
 * Add the multi selection widget
 * Add index to one2many's on_change

doc/ref/models/models.rst

@@ -406,6 +406,15 @@ Class methods:
     Return a sorted list of all revisions for ids. The list is composed of
     the date, id and username of the revision.
 
+.. classmethod:: ModelSQL.restore_history(ids, datetime)
+
+    Restore the record ids from history at the specified date time.
+    Restoring a record will still generate an entry in the history table.
+
+    .. warning::
+        No access rights are verified and the records are not validated.
+    ..
+
 .. classmethod:: ModelStorage.search(domain[, offset[, limit[, order[, count[, query]]]]])
 
     Return a list of records that match the :ref:`domain <topics-domain>` or

trytond/model/modelsql.py

@@ -303,6 +303,62 @@ class ModelSQL(ModelStorage):
                 cursor.execute(*history.insert(hcolumns,
                         [[id_, Now(), user] for id_ in sub_ids]))
 
+    @classmethod
+    def restore_history(cls, ids, datetime):
+        'Restore record ids from history at the date time'
+        if not cls._history:
+            return
+        transaction = Transaction()
+        cursor = transaction.cursor
+        in_max = cursor.IN_MAX
+        table = cls.__table__()
+        history = cls.__table_history__()
+        columns = []
+        hcolumns = []
+        for fname, field in sorted(cls._fields.iteritems()):
+            if hasattr(field, 'set'):
+                continue
+            columns.append(Column(table, fname))
+            if fname == 'write_uid':
+                hcolumns.append(Literal(transaction.user))
+            elif fname == 'write_date':
+                hcolumns.append(Now())
+            else:
+                hcolumns.append(Column(history, fname))
+
+        to_delete = []
+        to_update = []
+        for id_ in ids:
+            column_datetime = Coalesce(history.write_date, history.create_date)
+            hwhere = (column_datetime <= datetime) & (history.id == id_)
+            horder = column_datetime.desc
+            cursor.execute(*history.select(*hcolumns,
+                    where=hwhere, order_by=horder, limit=1))
+            values = cursor.fetchone()
+            if not values:
+                to_delete.append(id_)
+            else:
+                to_update.append(id_)
+                values = list(values)
+                cursor.execute(*table.update(columns, values,
+                        where=table.id == id_))
+                rowcount = cursor.rowcount
+                if rowcount == -1 or rowcount is None:
+                    cursor.execute(*table.select(table.id,
+                            where=table.id == id_))
+                    rowcount = len(cursor.fetchall())
+                if rowcount < 1:
+                    cursor.execute(*table.insert(columns, [values]))
+
+        if to_delete:
+            for i in range(0, len(to_delete), in_max):
+                sub_ids = to_delete[i:i + in_max]
+                where = reduce_ids(table.id, sub_ids)
+                cursor.execute(*table.delete(where=where))
+            cls.__insert_history(to_delete, True)
+        if to_update:
+            cls.__insert_history(to_update)
+
     @classmethod
     def __check_timestamp(cls, ids):
         transaction = Transaction()

trytond/tests/test_history.py

@@ -143,6 +143,36 @@ class HistoryTestCase(unittest.TestCase):
                     (first, history_id, u'Administrator'),
                     ])
 
+    def test0040restore_history(self):
+        'Test restore history'
+        History = POOL.get('test.history')
+
+        with Transaction().start(DB_NAME, USER,
+                context=CONTEXT) as transaction:
+            history = History(value=1)
+            history.save()
+            history_id = history.id
+            first = history.create_date
+
+            transaction.cursor.commit()
+
+        with Transaction().start(DB_NAME, USER,
+                context=CONTEXT) as transaction:
+            history = History(history_id)
+            history.value = 2
+            history.save()
+
+            transaction.cursor.commit()
+
+        with Transaction().start(DB_NAME, USER, context=CONTEXT):
+            History.restore_history([history_id], first)
+            history = History(history_id)
+            self.assertEqual(history.value, 1)
+
+        with Transaction().start(DB_NAME, USER, context=CONTEXT):
+            History.restore_history([history_id], datetime.datetime.min)
+            self.assertRaises(UserError, History.read, [history_id])
+
 
 def suite():
     return unittest.TestLoader().loadTestsFromTestCase(HistoryTestCase)