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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
import re
import json
MYPY = False
if MYPY:
# MYPY is set to True when run under Mypy.
from typing import Any, AnyStr, Callable, Dict, IO, Text
__all__ = ["load", "dump_local", "dump_local", "dump_dist", "dumps_dist"]
try:
import ujson
except ImportError:
has_ujson = False
else:
has_ujson = True
#
# load
#
if has_ujson:
load = ujson.load # type: Callable[[IO[AnyStr]], Any]
else:
load = json.load
#
# loads
#
if has_ujson:
loads = ujson.loads # type: Callable[[AnyStr], Any]
else:
loads = json.loads
#
# dump/dumps_local options for some libraries
#
_ujson_dump_local_kwargs = {
'ensure_ascii': False,
'escape_forward_slashes': False,
'indent': 1,
'reject_bytes': True,
} # type: Dict[str, Any]
_json_dump_local_kwargs = {
'ensure_ascii': False,
'indent': 1,
'separators': (',', ': '),
} # type: Dict[str, Any]
#
# dump_local (for local, non-distributed usage of JSON)
#
if has_ujson:
def dump_local(obj, fp):
# type: (Any, IO[str]) -> None
return ujson.dump(obj, fp, **_ujson_dump_local_kwargs)
else:
def dump_local(obj, fp):
# type: (Any, IO[str]) -> None
return json.dump(obj, fp, **_json_dump_local_kwargs)
#
# dumps_local (for local, non-distributed usage of JSON)
#
if has_ujson:
def dumps_local(obj):
# type: (Any) -> Text
return ujson.dumps(obj, **_ujson_dump_local_kwargs)
else:
def dumps_local(obj):
# type: (Any) -> Text
return json.dumps(obj, **_json_dump_local_kwargs)
#
# dump/dumps_dist (for distributed usage of JSON where files should safely roundtrip)
#
_ujson_dump_dist_kwargs = {
'sort_keys': True,
'indent': 1,
'reject_bytes': True,
} # type: Dict[str, Any]
_json_dump_dist_kwargs = {
'sort_keys': True,
'indent': 1,
'separators': (',', ': '),
} # type: Dict[str, Any]
if has_ujson:
if ujson.dumps([], indent=1) == "[]":
# optimistically see if https://github.com/ultrajson/ultrajson/issues/429 is fixed
def _ujson_fixup(s):
# type: (str) -> str
return s
else:
_ujson_fixup_re = re.compile(r"([\[{])[\n\x20]+([}\]])")
def _ujson_fixup(s):
# type: (str) -> str
return _ujson_fixup_re.sub(
lambda m: m.group(1) + m.group(2),
s
)
def dump_dist(obj, fp):
# type: (Any, IO[str]) -> None
fp.write(_ujson_fixup(ujson.dumps(obj, **_ujson_dump_dist_kwargs)))
def dumps_dist(obj):
# type: (Any) -> Text
return _ujson_fixup(ujson.dumps(obj, **_ujson_dump_dist_kwargs))
else:
def dump_dist(obj, fp):
# type: (Any, IO[str]) -> None
json.dump(obj, fp, **_json_dump_dist_kwargs)
def dumps_dist(obj):
# type: (Any) -> Text
return json.dumps(obj, **_json_dump_dist_kwargs)
|
