1. 模块列表
  2. 函数列表
  3. psycopg2._json()

Python psycopg2 模块，_json() 实例源码

我们从Python开源项目中，提取了以下16个代码示例，用于说明如何使用psycopg2._json()。

项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)

        def loads(s):
            return psycopg2.extras.json.loads(s, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assertTrue(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)

        def loads(s):
            return psycopg2.extras.json.loads(s, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：ShelbySearch    作者：Agentscreech    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)

        def loads(s):
            return psycopg2.extras.json.loads(s, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assertTrue(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：Price-Comparator    作者：Thejas-1    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：nmbs-realtime-feed    作者：datamindedbe    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)

        def loads(s):
            return psycopg2.extras.json.loads(s, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：aws-lambda-redshift-copy    作者：christianhxc    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：PyEloqua-Examples    作者：colemanja91    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)
        loads = lambda x: psycopg2.extras.json.loads(x, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assertTrue(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)
项目：flask    作者：bobohope    | 项目源码 | 文件源码 
def test_no_conn_curs(self):
        from psycopg2._json import _get_json_oids
        oid, array_oid = _get_json_oids(self.conn)

        old = psycopg2.extensions.string_types.get(114)
        olda = psycopg2.extensions.string_types.get(199)

        def loads(s):
            return psycopg2.extras.json.loads(s, parse_float=Decimal)
        try:
            new, newa = psycopg2.extras.register_json(
                loads=loads, oid=oid, array_oid=array_oid)
            curs = self.conn.cursor()
            curs.execute("""select '{"a": 100.0, "b": null}'::json""")
            data = curs.fetchone()[0]
            self.assert_(isinstance(data['a'], Decimal))
            self.assertEqual(data['a'], Decimal('100.0'))
        finally:
            psycopg2.extensions.string_types.pop(new.values[0])
            psycopg2.extensions.string_types.pop(newa.values[0])
            if old:
                psycopg2.extensions.register_type(old)
            if olda:
                psycopg2.extensions.register_type(olda)