diff --git a/IPython/kernel/zmq/session.py b/IPython/kernel/zmq/session.py index 5b71c36..50d29a4 100644 --- a/IPython/kernel/zmq/session.py +++ b/IPython/kernel/zmq/session.py @@ -81,9 +81,9 @@ 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 @@ 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 @@ 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 @@ 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]) diff --git a/IPython/kernel/zmq/tests/test_session.py b/IPython/kernel/zmq/tests/test_session.py index 06b6d6f..36c8d40 100644 --- a/IPython/kernel/zmq/tests/test_session.py +++ b/IPython/kernel/zmq/tests/test_session.py @@ -13,6 +13,8 @@ import os import uuid +from datetime import datetime + import zmq from zmq.tests import BaseZMQTestCase @@ -20,6 +22,10 @@ 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 @@ 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 @@ 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) diff --git a/IPython/parallel/client/client.py b/IPython/parallel/client/client.py index 3c37ac9..27e5d45 100644 --- a/IPython/parallel/client/client.py +++ b/IPython/parallel/client/client.py @@ -37,7 +37,7 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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] diff --git a/IPython/parallel/controller/hub.py b/IPython/parallel/controller/hub.py index 830fbc8..8bdba12 100644 --- a/IPython/parallel/controller/hub.py +++ b/IPython/parallel/controller/hub.py @@ -30,6 +30,7 @@ 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 @@ 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() diff --git a/IPython/utils/jsonutil.py b/IPython/utils/jsonutil.py index bcdcd73..565b1bc 100644 --- a/IPython/utils/jsonutil.py +++ b/IPython/utils/jsonutil.py @@ -33,8 +33,8 @@ 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 @@ 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): diff --git a/IPython/utils/tests/test_jsonutil.py b/IPython/utils/tests/test_jsonutil.py index 655f516..66bb6a4 100644 --- a/IPython/utils/tests/test_jsonutil.py +++ b/IPython/utils/tests/test_jsonutil.py @@ -113,6 +113,18 @@ 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)