python类Json()的实例源码

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 get_prep_value(self, value):
        if value is not None:
            return Json(value)
        return 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 get_prep_value(self, value):
        if value is not None:
            return Json(value)
        return value

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.assertQuotedEqual(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())

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'"))

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 test_module_not_available(self):
        from psycopg2.extras import Json
        self.assertRaises(ImportError, Json(None).getquoted)