forked from GetStream/stream-python
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathserializer.py
More file actions
54 lines (44 loc) · 1.8 KB
/
serializer.py
File metadata and controls
54 lines (44 loc) · 1.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import datetime
import json
import six
"""
Adds the ability to send date and datetime objects to the API
Datetime objects will be encoded/ decoded with microseconds
The date and datetime formats from the API are automatically supported and parsed
"""
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
DATE_FORMAT = "%Y-%m-%d"
def _datetime_encoder(obj):
if isinstance(obj, datetime.datetime):
return datetime.datetime.strftime(obj, DATETIME_FORMAT)
if isinstance(obj, datetime.date):
return datetime.datetime.strftime(obj, DATE_FORMAT)
def _datetime_decoder(dict_):
for key, value in dict_.items():
# The built-in `json` library will `unicode` strings, except for empty
# strings which are of type `str`. `jsondate` patches this for
# consistency so that `unicode` is always returned.
if value == "":
dict_[key] = ""
continue
if value is not None and isinstance(value, six.string_types):
try:
# The api always returns times like this
# 2014-07-25T09:12:24.735
datetime_obj = datetime.datetime.strptime(value, DATETIME_FORMAT)
dict_[key] = datetime_obj
except (ValueError, TypeError):
try:
# The api always returns times like this
# 2014-07-25T09:12:24.735
datetime_obj = datetime.datetime.strptime(value, DATE_FORMAT)
dict_[key] = datetime_obj.date()
except (ValueError, TypeError):
continue
return dict_
def dumps(*args, **kwargs):
kwargs["default"] = _datetime_encoder
return json.dumps(*args, **kwargs)
def loads(*args, **kwargs):
kwargs["object_hook"] = _datetime_decoder
return json.loads(*args, **kwargs)