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'")
python类Json()的实例源码
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_prep_lookup(self, lookup_type, value):
if lookup_type in ('has_key', 'has_keys', 'has_any_keys'):
return value
return (Json(dict(value), dumps=json_serialiser)
if isinstance(value, dict)
else super().get_prep_lookup(lookup_type, value))
# def validate(self, value, model_instance):
# super(JSONField, self).validate(value, model_instance)
# try:
# json.dumps(dict(value))
# except TypeError:
# raise exceptions.ValidationError(
# self.error_messages['invalid'],
# code='invalid',
# params={'value': value},
# )
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_scs(self):
cnn_on = self.connect(options="-c standard_conforming_strings=on")
cur_on = cnn_on.cursor()
self.assertEqual(
cur_on.mogrify("%s", [psycopg2.extras.Json({'a': '"'})]),
b'\'{"a": "\\""}\'')
cnn_off = self.connect(options="-c standard_conforming_strings=off")
cur_off = cnn_off.cursor()
self.assertEqual(
cur_off.mogrify("%s", [psycopg2.extras.Json({'a': '"'})]),
b'E\'{"a": "\\\\""}\'')
self.assertEqual(
cur_on.mogrify("%s", [psycopg2.extras.Json({'a': '"'})]),
b'\'{"a": "\\""}\'')
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_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_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'")
test_types_extras.py 文件源码
项目：aws-lambda-redshift-copy
作者: christianhxc
项目源码
文件源码
阅读 20
收藏 0
点赞 0
评论 0
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_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_scs(self):
cnn_on = self.connect(options="-c standard_conforming_strings=on")
cur_on = cnn_on.cursor()
self.assertEqual(
cur_on.mogrify("%s", [psycopg2.extras.Json({'a': '"'})]),
b'\'{"a": "\\""}\'')
cnn_off = self.connect(options="-c standard_conforming_strings=off")
cur_off = cnn_off.cursor()
self.assertEqual(
cur_off.mogrify("%s", [psycopg2.extras.Json({'a': '"'})]),
b'E\'{"a": "\\\\""}\'')
self.assertEqual(
cur_on.mogrify("%s", [psycopg2.extras.Json({'a': '"'})]),
b'\'{"a": "\\""}\'')
def update_metadata(cls, table, column, srid, scale_x, scale_y, scale_z,
offset_x, offset_y, offset_z):
'''
Add an entry to the lopocs metadata tables to use.
To be used after a fresh pc table creation.
'''
pcid = cls.query("""
select pcid from pointcloud_columns
where "schema" = %s and "table" = %s and "column" = %s
""", (table.split('.')[0], table.split('.')[1], column)
)[0][0]
bbox = cls.compute_boundingbox(table, column)
# compute bbox with offset and scale applied
bbox_scaled = [0] * 6
bbox_scaled[0] = (bbox['xmin'] - offset_x) / scale_x
bbox_scaled[1] = (bbox['ymin'] - offset_y) / scale_y
bbox_scaled[2] = (bbox['zmin'] - offset_z) / scale_z
bbox_scaled[3] = (bbox['xmax'] - offset_x) / scale_x
bbox_scaled[4] = (bbox['ymax'] - offset_y) / scale_y
bbox_scaled[5] = (bbox['zmax'] - offset_z) / scale_z
res = cls.query("""
delete from pointcloud_lopocs where schematable = %s and "column" = %s;
insert into pointcloud_lopocs (schematable, "column", srid, bbox)
values (%s, %s, %s, %s) returning id
""", (table, column, table, column, srid, bbox))
plid = res[0][0]
scales = scale_x, scale_y, scale_z
offsets = offset_x, offset_y, offset_z
json_schema = cls.patch2greyhoundschema(table, column)
cls.execute("""
insert into pointcloud_lopocs_outputs
(id, pcid, scales, offsets, stored, bbox, point_schema)
values (%s, %s, %s, %s, True, %s, %s)
""", (
plid, pcid, iterable2pgarray(scales), iterable2pgarray(offsets),
iterable2pgarray(bbox_scaled), Json(json_schema)))
def test_module_not_available(self):
from psycopg2.extras import Json
self.assertRaises(ImportError, Json(None).getquoted)
def test_customizable_with_module_not_available(self):
from psycopg2.extras import Json
class MyJson(Json):
def dumps(self, obj):
assert obj is None
return "hi"
self.assertEqual(MyJson(None).getquoted(), "'hi'")
def test_adapt(self):
from psycopg2.extras import json, Json
objs = [None, "te'xt", 123, 123.45,
'\xe0\u20ac', ['a', 100], {'a': 100}]
curs = self.conn.cursor()
for obj in enumerate(objs):
self.assertEqual(curs.mogrify("%s", (Json(obj),)),
psycopg2.extensions.QuotedString(json.dumps(obj)).getquoted())
def test_str(self):
snowman = "\u2603"
obj = {'a': [1, 2, snowman]}
j = psycopg2.extensions.adapt(psycopg2.extras.Json(obj))
s = str(j)
self.assertTrue(isinstance(s, str))
# no pesky b's
self.assertTrue(s.startswith("'"))
self.assertTrue(s.endswith("'"))
def test_module_not_available(self):
from psycopg2.extras import Json
self.assertRaises(ImportError, Json(None).getquoted)
def test_customizable_with_module_not_available(self):
from psycopg2.extras import Json
class MyJson(Json):
def dumps(self, obj):
assert obj is None
return "hi"
self.assertEqual(MyJson(None).getquoted(), "'hi'")
def test_adapt(self):
from psycopg2.extras import json, Json
objs = [None, "te'xt", 123, 123.45,
u'\xe0\u20ac', ['a', 100], {'a': 100}]
curs = self.conn.cursor()
for obj in enumerate(objs):
self.assertEqual(curs.mogrify("%s", (Json(obj),)),
psycopg2.extensions.QuotedString(json.dumps(obj)).getquoted())
