Merge pull request #4497 from minrk/datetime-session

don't automatically unpack datetime objects in the message spec

only unpack in headers
move handling of datetime objects in content and metadata to application-level code
minor fix to ISO8601 re (closes  #4493)

IPython/kernel/zmq/session.py

@@ -81,9 +81,9 b' def squash_unicode(obj):'
 
 # ISO8601-ify datetime objects
 json_packer = lambda obj: jsonapi.dumps(obj, default=date_default)
-json_unpacker = lambda s: extract_dates(jsonapi.loads(s))
+json_unpacker = lambda s: jsonapi.loads(s)
 
-pickle_packer = lambda o: pickle.dumps(o,-1)
+pickle_packer = lambda o: pickle.dumps(squash_dates(o),-1)
 pickle_unpacker = pickle.loads
 
 default_packer = json_packer
@@ -429,7 +429,7 b' class Session(Configurable):'
         return str(uuid.uuid4())
 
     def _check_packers(self):
-        """check packers for binary data and datetime support."""
+        """check packers for datetime support."""
         pack = self.pack
         unpack = self.unpack
 
@@ -469,9 +469,11 b' class Session(Configurable):'
         msg = dict(t=datetime.now())
         try:
             unpacked = unpack(pack(msg))
+            if isinstance(unpacked['t'], datetime):
+                raise ValueError("Shouldn't deserialize to datetime")
         except Exception:
             self.pack = lambda o: pack(squash_dates(o))
-            self.unpack = lambda s: extract_dates(unpack(s))
+            self.unpack = lambda s: unpack(s)
 
     def msg_header(self, msg_type):
         return msg_header(self.msg_id, msg_type, self.username, self.session)
@@ -815,10 +817,10 b' class Session(Configurable):'
         if not len(msg_list) >= minlen:
             raise TypeError("malformed message, must have at least %i elements"%minlen)
         header = self.unpack(msg_list[1])
-        message['header'] = header
+        message['header'] = extract_dates(header)
         message['msg_id'] = header['msg_id']
         message['msg_type'] = header['msg_type']
-        message['parent_header'] = self.unpack(msg_list[2])
+        message['parent_header'] = extract_dates(self.unpack(msg_list[2]))
         message['metadata'] = self.unpack(msg_list[3])
         if content:
             message['content'] = self.unpack(msg_list[4])

IPython/kernel/zmq/tests/test_session.py

@@ -13,6 +13,8 b''
 
 import os
 import uuid
+from datetime import datetime
+
 import zmq
 
 from zmq.tests import BaseZMQTestCase
@@ -20,6 +22,10 b' from zmq.eventloop.zmqstream import ZMQStream'
 
 from IPython.kernel.zmq import session as ss
 
+from IPython.testing.decorators import skipif, module_not_available
+from IPython.utils.py3compat import string_types
+from IPython.utils import jsonutil
+
 def _bad_packer(obj):
     raise TypeError("I don't work")
 
@@ -155,24 +161,6 b' class TestSession(SessionTestCase):'
         t.wait(1) # this will raise
 
 
-    # def test_rekey(self):
-    #     """rekeying dict around json str keys"""
-    #     d = {'0': uuid.uuid4(), 0:uuid.uuid4()}
-    #     self.assertRaises(KeyError, ss.rekey, d)
-    #
-    #     d = {'0': uuid.uuid4(), 1:uuid.uuid4(), 'asdf':uuid.uuid4()}
-    #     d2 = {0:d['0'],1:d[1],'asdf':d['asdf']}
-    #     rd = ss.rekey(d)
-    #     self.assertEqual(d2,rd)
-    #
-    #     d = {'1.5':uuid.uuid4(),'1':uuid.uuid4()}
-    #     d2 = {1.5:d['1.5'],1:d['1']}
-    #     rd = ss.rekey(d)
-    #     self.assertEqual(d2,rd)
-    #
-    #     d = {'1.0':uuid.uuid4(),'1':uuid.uuid4()}
-    #     self.assertRaises(KeyError, ss.rekey, d)
-    #
     def test_unique_msg_ids(self):
         """test that messages receive unique ids"""
         ids = set()
@@ -267,5 +255,35 b' class TestSession(SessionTestCase):'
     
     def test_bad_roundtrip(self):
         with self.assertRaises(ValueError):
-            session=  ss.Session(unpack=lambda b: 5)
+            session = ss.Session(unpack=lambda b: 5)
+    
+    def _datetime_test(self, session):
+        content = dict(t=datetime.now())
+        metadata = dict(t=datetime.now())
+        p = session.msg('msg')
+        msg = session.msg('msg', content=content, metadata=metadata, parent=p['header'])
+        smsg = session.serialize(msg)
+        msg2 = session.unserialize(session.feed_identities(smsg)[1])
+        assert isinstance(msg2['header']['date'], datetime)
+        self.assertEqual(msg['header'], msg2['header'])
+        self.assertEqual(msg['parent_header'], msg2['parent_header'])
+        self.assertEqual(msg['parent_header'], msg2['parent_header'])
+        assert isinstance(msg['content']['t'], datetime)
+        assert isinstance(msg['metadata']['t'], datetime)
+        assert isinstance(msg2['content']['t'], string_types)
+        assert isinstance(msg2['metadata']['t'], string_types)
+        self.assertEqual(msg['content'], jsonutil.extract_dates(msg2['content']))
+        self.assertEqual(msg['content'], jsonutil.extract_dates(msg2['content']))
+    
+    def test_datetimes(self):
+        self._datetime_test(self.session)
+    
+    def test_datetimes_pickle(self):
+        session = ss.Session(packer='pickle')
+        self._datetime_test(session)
+    
+    @skipif(module_not_available('msgpack'))
+    def test_datetimes_msgpack(self):
+        session = ss.Session(packer='msgpack.packb', unpacker='msgpack.unpackb')
+        self._datetime_test(session)
     

IPython/parallel/client/client.py

@@ -37,7 +37,7 b' from IPython.core.profiledir import ProfileDir, ProfileDirError'
 
 from IPython.utils.capture import RichOutput
 from IPython.utils.coloransi import TermColors
-from IPython.utils.jsonutil import rekey
+from IPython.utils.jsonutil import rekey, extract_dates, parse_date
 from IPython.utils.localinterfaces import localhost, is_local_ip
 from IPython.utils.path import get_ipython_dir
 from IPython.utils.py3compat import cast_bytes, string_types, xrange, iteritems
@@ -675,7 +675,7 b' class Client(HasTraits):'
         if 'date' in parent:
             md['submitted'] = parent['date']
         if 'started' in msg_meta:
-            md['started'] = msg_meta['started']
+            md['started'] = parse_date(msg_meta['started'])
         if 'date' in header:
             md['completed'] = header['date']
         return md
@@ -1564,8 +1564,8 b' class Client(HasTraits):'
         for msg_id in sorted(theids):
             if msg_id in content['completed']:
                 rec = content[msg_id]
-                parent = rec['header']
-                header = rec['result_header']
+                parent = extract_dates(rec['header'])
+                header = extract_dates(rec['result_header'])
                 rcontent = rec['result_content']
                 iodict = rec['io']
                 if isinstance(rcontent, str):
@@ -1580,7 +1580,7 b' class Client(HasTraits):'
                 )
                 md.update(self._extract_metadata(md_msg))
                 if rec.get('received'):
-                    md['received'] = rec['received']
+                    md['received'] = parse_date(rec['received'])
                 md.update(iodict)
                 
                 if rcontent['status'] == 'ok':
@@ -1842,6 +1842,13 b' class Client(HasTraits):'
         has_bufs = buffer_lens is not None
         has_rbufs = result_buffer_lens is not None
         for i,rec in enumerate(records):
+            # unpack datetime objects
+            for hkey in ('header', 'result_header'):
+                if hkey in rec:
+                    rec[hkey] = extract_dates(rec[hkey])
+            for dtkey in ('submitted', 'started', 'completed', 'received'):
+                if dtkey in rec:
+                    rec[dtkey] = parse_date(rec[dtkey])
             # relink buffers
             if has_bufs:
                 blen = buffer_lens[i]

IPython/parallel/controller/hub.py

@@ -30,6 +30,7 b' from zmq.eventloop.zmqstream import ZMQStream'
 
 # internal:
 from IPython.utils.importstring import import_item
+from IPython.utils.jsonutil import extract_dates
 from IPython.utils.localinterfaces import localhost
 from IPython.utils.py3compat import cast_bytes, unicode_type, iteritems
 from IPython.utils.traitlets import (
@@ -1385,7 +1386,7 b' class Hub(SessionFactory):'
     def db_query(self, client_id, msg):
         """Perform a raw query on the task record database."""
         content = msg['content']
-        query = content.get('query', {})
+        query = extract_dates(content.get('query', {}))
         keys = content.get('keys', None)
         buffers = []
         empty = list()

IPython/utils/jsonutil.py

@@ -33,8 +33,8 b" next_attr_name = '__next__' if py3compat.PY3 else 'next'"
 #-----------------------------------------------------------------------------
 
 # timestamp formats
-ISO8601="%Y-%m-%dT%H:%M:%S.%f"
-ISO8601_PAT=re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+)Z?([\+\-]\d{2}:?\d{2})?$")
+ISO8601 = "%Y-%m-%dT%H:%M:%S.%f"
+ISO8601_PAT=re.compile(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{1,6})Z?([\+\-]\d{2}:?\d{2})?$")
 
 #-----------------------------------------------------------------------------
 # Classes and functions
@@ -62,22 +62,34 b' def rekey(dikt):'
             dikt[nk] = dikt.pop(k)
     return dikt
 
+def parse_date(s):
+    """parse an ISO8601 date string
+    
+    If it is None or not a valid ISO8601 timestamp,
+    it will be returned unmodified.
+    Otherwise, it will return a datetime object.
+    """
+    if s is None:
+        return s
+    m = ISO8601_PAT.match(s)
+    if m:
+        # FIXME: add actual timezone support
+        # this just drops the timezone info
+        notz = m.groups()[0]
+        return datetime.strptime(notz, ISO8601)
+    return s
 
 def extract_dates(obj):
     """extract ISO8601 dates from unpacked JSON"""
     if isinstance(obj, dict):
-        obj = dict(obj) # don't clobber
+        new_obj = {} # don't clobber
         for k,v in iteritems(obj):
-            obj[k] = extract_dates(v)
+            new_obj[k] = extract_dates(v)
+        obj = new_obj
     elif isinstance(obj, (list, tuple)):
         obj = [ extract_dates(o) for o in obj ]
     elif isinstance(obj, string_types):
-        m = ISO8601_PAT.match(obj)
-        if m:
-            # FIXME: add actual timezone support
-            # this just drops the timezone info
-            notz = m.groups()[0]
-            obj = datetime.strptime(notz, ISO8601)
+        obj = parse_date(obj)
     return obj
 
 def squash_dates(obj):

IPython/utils/tests/test_jsonutil.py

@@ -113,6 +113,18 b' def test_extract_dates():'
         nt.assert_true(isinstance(dt, datetime.datetime))
         nt.assert_equal(dt, ref)
 
+def test_parse_ms_precision():
+    base = '2013-07-03T16:34:52.'
+    digits = '1234567890'
+    
+    for i in range(len(digits)):
+        ts = base + digits[:i]
+        parsed = jsonutil.parse_date(ts)
+        if i >= 1 and i <= 6:
+            assert isinstance(parsed, datetime.datetime)
+        else:
+            assert isinstance(parsed, str)
+
 def test_date_default():
     data = dict(today=datetime.datetime.now(), utcnow=tz.utcnow())
     jsondata = json.dumps(data, default=jsonutil.date_default)