use 'timestamp' datatype for timestamps in sqlitedb

IPython/parallel/controller/dictdb.py

@@ -46,7 +46,7 b' from datetime import datetime'
 
 from IPython.config.configurable import Configurable
 
-from IPython.utils.traitlets import Dict, Unicode
+from IPython.utils.traitlets import Dict, Unicode, Instance
 
 filters = {
  '$lt' : lambda a,b: a < b,
@@ -83,6 +83,7 b' class BaseDB(Configurable):'
     """Empty Parent class so traitlets work on DB."""
     # base configurable traits:
     session = Unicode("")
+    log = Instance('logging.Logger', ('root',))
 
 class DictDB(BaseDB):
     """Basic in-memory dict-based object for saving Task Records.

IPython/parallel/controller/sqlitedb.py

@@ -15,9 +15,9 b' import sqlite3'
 
 from zmq.eventloop import ioloop
 
-from IPython.utils.traitlets import Unicode, Instance, List
+from IPython.utils.traitlets import Unicode, Instance, List, Dict
 from .dictdb import BaseDB
-from IPython.utils.jsonutil import date_default, extract_dates
+from IPython.utils.jsonutil import date_default, extract_dates, squash_dates
 
 #-----------------------------------------------------------------------------
 # SQLite operators, adapters, and converters
@@ -42,15 +42,6 b' null_operators = {'
 '!=' : "IS NOT NULL",
 }
 
-def _adapt_datetime(dt):
-    return dt.strftime(ISO8601)
-
-def _convert_datetime(ds):
-    if ds is None:
-        return ds
-    else:
-        return datetime.strptime(ds, ISO8601)
-
 def _adapt_dict(d):
     return json.dumps(d, default=date_default)
 
@@ -95,6 +86,7 b' class SQLiteDB(BaseDB):'
         get_result methods.""")
     
     _db = Instance('sqlite3.Connection')
+    # the ordered list of column names
     _keys = List(['msg_id' ,
             'header' ,
             'content',
@@ -115,6 +107,27 b' class SQLiteDB(BaseDB):'
             'stdout',
             'stderr',
         ])
+    # sqlite datatypes for checking that db is current format
+    _types = Dict({'msg_id' : 'text' ,
+            'header' : 'dict text',
+            'content' : 'dict text',
+            'buffers' : 'bufs blob',
+            'submitted' : 'timestamp',
+            'client_uuid' : 'text',
+            'engine_uuid' : 'text',
+            'started' : 'timestamp',
+            'completed' : 'timestamp',
+            'resubmitted' : 'timestamp',
+            'result_header' : 'dict text',
+            'result_content' : 'dict text',
+            'result_buffers' : 'bufs blob',
+            'queue' : 'text',
+            'pyin' : 'text',
+            'pyout' : 'text',
+            'pyerr' : 'text',
+            'stdout' : 'text',
+            'stderr' : 'text',
+        })
     
     def __init__(self, **kwargs):
         super(SQLiteDB, self).__init__(**kwargs)
@@ -149,11 +162,36 b' class SQLiteDB(BaseDB):'
             d[key] = None
         return d
     
+    def _check_table(self):
+        """Ensure that an incorrect table doesn't exist
+        
+        If a bad (old) table does exist, return False
+        """
+        cursor = self._db.execute("PRAGMA table_info(%s)"%self.table)
+        lines = cursor.fetchall()
+        if not lines:
+            # table does not exist
+            return True
+        types = {}
+        keys = []
+        for line in lines:
+            keys.append(line[1])
+            types[line[1]] = line[2]
+        if self._keys != keys:
+            # key mismatch
+            self.log.warn('keys mismatch')
+            return False
+        for key in self._keys:
+            if types[key] != self._types[key]:
+                self.log.warn(
+                    'type mismatch: %s: %s != %s'%(key,types[key],self._types[key])
+                )
+                return False
+        return True
+        
     def _init_db(self):
         """Connect to the database and get new session number."""
         # register adapters
-        sqlite3.register_adapter(datetime, _adapt_datetime)
-        sqlite3.register_converter('datetime', _convert_datetime)
         sqlite3.register_adapter(dict, _adapt_dict)
         sqlite3.register_converter('dict', _convert_dict)
         sqlite3.register_adapter(list, _adapt_bufs)
@@ -164,18 +202,27 b' class SQLiteDB(BaseDB):'
             # isolation_level = None)#,
              cached_statements=64)
         # print dir(self._db)
+        first_table = self.table
+        i=0
+        while not self._check_table():
+            i+=1
+            self.table = first_table+'_%i'%i
+            self.log.warn(
+                "Table %s exists and doesn't match db format, trying %s"%
+                (first_table,self.table)
+            )
         
         self._db.execute("""CREATE TABLE IF NOT EXISTS %s 
                 (msg_id text PRIMARY KEY,
                 header dict text,
                 content dict text,
                 buffers bufs blob,
-                submitted datetime text,
+                submitted timestamp,
                 client_uuid text,
                 engine_uuid text,
-                started datetime text,
-                completed datetime text,
-                resubmitted datetime text,
+                started timestamp,
+                completed timestamp,
+                resubmitted timestamp,
                 result_header dict text,
                 result_content dict text,
                 result_buffers bufs blob,