def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)), b"'123.45'")
python类JSONEncoder()的实例源码
def test_adapt_dumps(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
curs = self.conn.cursor()
obj = Decimal('123.45')
def dumps(obj):
return json.dumps(obj, cls=DecimalEncoder)
self.assertEqual(curs.mogrify("%s", (Json(obj, dumps=dumps),)),
b"'123.45'")
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)), b"'123.45'")
def test_adapt_dumps(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
curs = self.conn.cursor()
obj = Decimal('123.45')
def dumps(obj):
return json.dumps(obj, cls=DecimalEncoder)
self.assertEqual(curs.mogrify("%s", (Json(obj, dumps=dumps),)),
b"'123.45'")
def get_status(self):
jsonString = json.JSONEncoder().encode(
[ { "name":"RSU", "value": self.name},
{ "name": self.prefix+" Spaces_available","value": self.spaces_available},
{ "name": self.prefix+" Longitude","value":self.lon},
{ "name": self.prefix+" Latitude","value":self.lat},
{ "name": self.prefix+" Cost/hour","value":self.rate},
{ "name":"Count", "value":self.count}])
self.spaces_available = random.randrange(0,self.max_spaces,1);
self.rate = random.randrange(0,20,1);
self.count += 1
data = json.dumps(jsonString)
tdata = (data.replace("}, {","}\\\0{")).replace("\\","")
# LOG.info("TDATA = " + tdata)
return(tdata.replace("\"[","")).replace("]\"","")+chr(0) # Adding a \0 to the end of the packet per the message format standards
def write_json(output, end='', raw=False, file=None, flush=False):
file = file or sys.stdout
if len(output) == 1:
output = output[0]
if raw:
json.dump(output, file, separators=(',', ':'), cls=JSONEncoder)
else:
json.dump(output, file, indent=4, sort_keys=True, cls=JSONEncoder)
if flush:
file.flush()
if end:
write_output(file, '', end=end, sep='', flush=flush)
def default(self, o):
# for Enum Type
if isinstance(o, enum.Enum):
return o.value
# for Enum Select Integer
if isinstance(o, EnumInt):
return o.key
if isinstance(o, (datetime, date)):
return o.isoformat()
if isinstance(o, Decimal):
return _number_str(o)
if isinstance(o, ObjectId):
return str(o)
return super(JSONEncoder, self).default(o)
def to_json(self):
"""Serializes event sequences using json.
Returns
-------
str
json serialization of self.
"""
class EventEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Event):
return [obj.event_type, obj.event_time]
if isinstance(obj, EventSequence):
return obj.sequence_of_events
return json.JSONEncoder.default(self, obj)
return json.dumps(self, ensure_ascii=False, cls=EventEncoder)
def default(self, obj):
if isinstance(obj, complex):
return [obj.real, obj.imag]
elif isinstance(obj, datetime):
return obj.isoformat()
elif isinstance(obj, type):
return obj.__module__ + '.' + obj.__name__
elif isinstance(obj, InterfaceClass):
return [x.__module__ + '.' + x.__name__ for x in obj.__iro__] # noqa
try:
iterable = iter(obj)
except TypeError:
pass
else:
return list(iterable)
if isinstance(obj, PermissionSetting):
return obj.getName()
if callable(obj):
return obj.__module__ + '.' + obj.__name__
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, obj)
def to_json(o):
import json
from datetime import date
from datetime import datetime
class CJsonEncoder(json.JSONEncoder):
def __init__(self, **kwargs):
super(CJsonEncoder, self).__init__(**kwargs)
def default(self, obj):
if isinstance(obj, datetime):
return obj.strftime("%Y-%m-%d %H:%M:%S")
elif isinstance(obj, date):
return obj.strftime("%Y-%m-%d")
else:
return super(CJsonEncoder, self).default(obj)
return json.dumps(o, cls=CJsonEncoder, ensure_ascii=False)
def default(self, o):
if isinstance(o, datetime.datetime):
return o.isoformat()
if isinstance(o, datetime.date):
return o.isoformat()
if isinstance(o, datetime.time):
return o.isoformat()
if isinstance(o, decimal.Decimal):
return str(o)
return json.JSONEncoder.default(self, o)
# ????? json.dumps(yourobj, cls=DateTimeEncoder)
# class DecimalEncoder(json.JSONEncoder):
# def _iterencode(self, o, markers=None):
# if isinstance(o, decimal.Decimal):
# # wanted a simple yield str(o) in the next line,
# # but that would mean a yield on the line with super(...),
# # which wouldn't work (see my comment below), so...
# return (str(o) for o in [o])
# return super(DecimalEncoder, self)._iterencode(o, markers)
def default(self, obj):
if isinstance(obj, Sample):
sample_dict = dict(obj.get_dict())
# get sample dict and make a copy of it
sample_dict.pop("sampleProject")
if "sequencerSampleId" in sample_dict:
# if the sample ID field is populated, then we've just Finished
# reading the run from disk and we're preparing to send data
# to the server. The server is using the sample ID field as the
# name of the sample, so overwrite whatever we *were* using to
# find files with the sample ID field.
sample_dict["sampleName"] = sample_dict["sequencerSampleId"]
return sample_dict
else:
return json.JSONEncoder.default(self, obj)
def default(self, obj):
if isinstance(obj, complex):
return [obj.real, obj.imag]
elif isinstance(obj, datetime):
return obj.isoformat()
elif isinstance(obj, type):
return obj.__module__ + '.' + obj.__name__
elif isinstance(obj, InterfaceClass):
return [x.__module__ + '.' + x.__name__ for x in obj.__iro__] # noqa
try:
iterable = iter(obj)
except TypeError:
pass
else:
return list(iterable)
if isinstance(obj, PermissionSetting):
return obj.get_name()
if callable(obj):
return obj.__module__ + '.' + obj.__name__
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, obj)
# b/w compat import
def default(self, obj):
if hasattr(obj, '__json__'):
return obj.__json__()
elif isinstance(obj, collections.Iterable):
return list(obj)
elif isinstance(obj, datetime):
return obj.isoformat()
elif hasattr(obj, '__getitem__') and hasattr(obj, 'keys'):
return dict(obj)
elif hasattr(obj, '__dict__'):
return {member: getattr(obj, member)
for member in dir(obj)
if not member.startswith('_') and
not hasattr(getattr(obj, member), '__call__')}
return json.JSONEncoder.default(self, obj)
def to_json(o):
import json
from datetime import date
from datetime import datetime
class CJsonEncoder(json.JSONEncoder):
def __init__(self, **kwargs):
super(CJsonEncoder, self).__init__(**kwargs)
def default(self, obj):
if isinstance(obj, datetime):
return obj.strftime("%Y-%m-%d %H:%M:%S")
elif isinstance(obj, date):
return obj.strftime("%Y-%m-%d")
else:
return super(CJsonEncoder, self).default(obj)
return json.dumps(o, cls=CJsonEncoder, ensure_ascii=False)
def default(self, obj):
# convert dates and numpy objects in a json serializable format
if isinstance(obj, datetime):
return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
elif isinstance(obj, date):
return obj.strftime('%Y-%m-%d')
elif type(obj) in (np.int_, np.intc, np.intp, np.int8, np.int16,
np.int32, np.int64, np.uint8, np.uint16,
np.uint32, np.uint64):
return int(obj)
elif type(obj) in (np.bool_,):
return bool(obj)
elif type(obj) in (np.float_, np.float16, np.float32, np.float64,
np.complex_, np.complex64, np.complex128):
return float(obj)
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, obj)
def default(self, obj):
if type(obj) is value.ValuePermission:
return obj.value
if type(obj) is value.ValueStatus:
return obj.value
if type(obj) is state.StateType:
return obj.value
if type(obj) is state.StateStatus:
return obj.value
if type(obj) is value.ValueNumberType:
return obj.__dict__
if type(obj) is value.ValueStringType:
return obj.__dict__
if type(obj) is status.StatusType:
return obj.value
if type(obj) is status.StatusLevel:
return obj.value
# if type(obj) is value.ValueSetType:
# return obj.__dict__
# if type(obj) is value.ValueBlobType:
# return obj.__dict__
# if type(obj) is value.ValueXmlType:
# return obj.__dict__
# return json.JSONEncoder.default(self, obj)
def test_adapt_dumps(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
curs = self.conn.cursor()
obj = Decimal('123.45')
def dumps(obj):
return json.dumps(obj, cls=DecimalEncoder)
self.assertQuotedEqual(curs.mogrify("%s", (Json(obj, dumps=dumps),)),
b"'123.45'")
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertQuotedEqual(curs.mogrify("%s", (MyJson(obj),)), b"'123.45'")
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)),
b("'123.45'"))
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)),
b("'123.45'"))
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)),
b("'123.45'"))
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)),
b("'123.45'"))
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)),
b("'123.45'"))
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)),
b("'123.45'"))
def test_call_custom_encoder(self):
# Custom encoder for dates
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return json.JSONEncoder.default(self, obj)
now = datetime.now()
js_func = """var f = function(args) {
return args;
}"""
self.mr.eval(js_func)
self.assertEqual(self.mr.call('f', now, encoder=CustomEncoder),
now.isoformat())
def default(self, obj):
if isinstance(obj, MriTransform):
return {'name':obj.name,
'xfm' :obj.xfm,
'xfm_f':obj.xfm_f,
'xfm_inv' :obj.xfm_inv,
'xfm_f_inv':obj.xfm_f_inv,
'prefix':obj.prefix
}
elif isinstance(obj, MriDataset):
return {'name':obj.name,
'scan':obj.scan,
'mask':obj.mask,
'scan_f':obj.scan_f,
'mask_f':obj.mask_f,
'prefix':obj.prefix,
'add':obj.add,
'add_f':obj.add_f,
'group':obj.group,
'grading':obj.grading,
}
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, obj)
# kate: space-indent on; indent-width 4; indent-mode python;replace-tabs on;word-wrap-column 80;show-tabs on
def default(self, obj):
if isinstance(obj, MriTransform):
return {'name':obj.name,
'xfm' :obj.xfm,
'xfm_f':obj.xfm_f,
'xfm_inv' :obj.xfm_inv,
'xfm_f_inv':obj.xfm_f_inv,
'prefix':obj.prefix
}
elif isinstance(obj, MriDataset):
return {'name':obj.name,
'scan':obj.scan,
'mask':obj.mask,
'scan_f':obj.scan_f,
'mask_f':obj.mask_f,
'prefix':obj.prefix,
'add':obj.add,
'add_f':obj.add_f
}
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, obj)
# kate: space-indent on; indent-width 4; indent-mode python;replace-tabs on;word-wrap-column 80;show-tabs on
def __save_msl_data(self):
"""
Saves the keys and tokens in json file
:return:
"""
data = {
"encryption_key": base64.standard_b64encode(self.encryption_key),
'sign_key': base64.standard_b64encode(self.sign_key),
'tokens': {
'mastertoken': self.mastertoken
}
}
serialized_data = json.JSONEncoder().encode(data)
self.save_file(
msl_data_path=self.kodi_helper.msl_data_path,
filename='msl_data.json',
content=serialized_data)
def test_adapt_subclass(self):
from psycopg2.extras import json, Json
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
class MyJson(Json):
def dumps(self, obj):
return json.dumps(obj, cls=DecimalEncoder)
curs = self.conn.cursor()
obj = Decimal('123.45')
self.assertEqual(curs.mogrify("%s", (MyJson(obj),)),
b("'123.45'"))
