我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用psycopg2.extras.Json()。
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 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 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_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())
def test_register_on_dict(self): from psycopg2.extras import Json psycopg2.extensions.register_adapter(dict, Json) try: curs = self.conn.cursor() obj = {'a': 123} self.assertEqual(curs.mogrify("%s", (obj,)), b"""'{"a": 123}'""") finally: del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
def test_str(self): snowman = u"\u2603" obj = {'a': [1, 2, snowman]} j = psycopg2.extensions.adapt(psycopg2.extras.Json(obj)) s = str(j) self.assert_(isinstance(s, str)) # no pesky b's self.assert_(s.startswith("'")) self.assert_(s.endswith("'"))
def get_prep_value(self, value): if value is not None: return Json(value) return value
def get_prep_lookup(self, lookup_type, value): if lookup_type in ('has_key', 'has_keys', 'has_any_keys'): return value if isinstance(value, (dict, list)): return Json(value) return super(JSONField, self).get_prep_lookup(lookup_type, value)
def UpsertAggregateInstance(self, aggregateId, snapshot, indexValues): upsertStatement = PostgresStatement.UpsertAggregateInstance.format(self.tableNames.instanceBaseTableName) self.ExecuteSingle(upsertStatement, aggregateId, Json(indexValues), psycopg2.Binary(snapshot))
def UpdateEventPersistenceCheckpoint(self, requestId, indexValues, serializedResponse): insertStatement = PostgresStatement.InsertCheckpoint.format(self.tableNames.eventPersistenceStagingTableName) self.ExecuteSingle(insertStatement, requestId, Json(indexValues), psycopg2.Binary(serializedResponse))
def schema_update_conflicts(instance): if not hasattr(instance, '_attr_field'): raise ValueError("instance doesn't have an attribute field") conflicts = [] try: instance._attr_field._pre_save_selector_check(strict=True) except SchemaUpdateException as exc_info: conflicts = exc_info.conflicts return conflicts # This is needed to provide JSON serialisation for date objects # whenever they're saved to JSON attribute fields. This function is # passed as the custom "dumps" method for psycopg2's Json class to # use.
def get_prep_value(self, value): if value is None: return None return Json(dict(value), dumps=json_serialiser)
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.assertQuotedEqual(curs.mogrify("%s", (Json(obj),)), psycopg2.extensions.QuotedString(json.dumps(obj)).getquoted())
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())
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') dumps = lambda obj: json.dumps(obj, cls=DecimalEncoder) self.assertEqual(curs.mogrify("%s", (Json(obj, dumps=dumps),)), b("'123.45'"))
