我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用psycopg2.extensions.adapters()。
def test_none_fast_path(self): # the None adapter is not actually invoked in regular adaptation class WonkyAdapter(object): def __init__(self, obj): pass def getquoted(self): return "NOPE!" curs = self.conn.cursor() orig_adapter = ext.adapters[type(None), ext.ISQLQuote] try: ext.register_adapter(type(None), WonkyAdapter) self.assertEqual(ext.adapt(None).getquoted(), "NOPE!") s = curs.mogrify("SELECT %s;", (None,)) self.assertEqual(b"SELECT NULL;", s) finally: ext.register_adapter(type(None), orig_adapter)
def test_none_fast_path(self): # the None adapter is not actually invoked in regular adaptation class WonkyAdapter(object): def __init__(self, obj): pass def getquoted(self): return "NOPE!" curs = self.conn.cursor() orig_adapter = ext.adapters[type(None), ext.ISQLQuote] try: ext.register_adapter(type(None), WonkyAdapter) self.assertEqual(ext.adapt(None).getquoted(), "NOPE!") s = curs.mogrify("SELECT %s;", (None,)) self.assertEqual(b("SELECT NULL;"), s) finally: ext.register_adapter(type(None), orig_adapter)
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_register_range_adapter(self): from psycopg2.extras import Range, register_range cur = self.conn.cursor() cur.execute("create type textrange as range (subtype=text)") rc = register_range('textrange', 'TextRange', cur) TextRange = rc.range self.assertTrue(issubclass(TextRange, Range)) self.assertEqual(TextRange.__name__, 'TextRange') r = TextRange('a', 'b', '(]') cur.execute("select %s", (r,)) r1 = cur.fetchone()[0] self.assertEqual(r1.lower, 'a') self.assertEqual(r1.upper, 'b') self.assertTrue(not r1.lower_inc) self.assertTrue(r1.upper_inc) cur.execute("select %s", ([r, r, r],)) rs = cur.fetchone()[0] self.assertEqual(len(rs), 3) for r1 in rs: self.assertEqual(r1.lower, 'a') self.assertEqual(r1.upper, 'b') self.assertTrue(not r1.lower_inc) self.assertTrue(r1.upper_inc) # clear the adapters to allow precise count by scripts/refcounter.py del ext.adapters[rc.range, ext.ISQLQuote]
def test_schema_range(self): cur = self.conn.cursor() cur.execute("create schema rs") cur.execute("create type r1 as range (subtype=text)") cur.execute("create type r2 as range (subtype=text)") cur.execute("create type rs.r2 as range (subtype=text)") cur.execute("create type rs.r3 as range (subtype=text)") cur.execute("savepoint x") from psycopg2.extras import register_range ra1 = register_range('r1', 'r1', cur) ra2 = register_range('r2', 'r2', cur) rars2 = register_range('rs.r2', 'r2', cur) rars3 = register_range('rs.r3', 'r3', cur) self.assertNotEqual( ra2.typecaster.values[0], rars2.typecaster.values[0]) self.assertRaises(psycopg2.ProgrammingError, register_range, 'r3', 'FailRange', cur) cur.execute("rollback to savepoint x;") self.assertRaises(psycopg2.ProgrammingError, register_range, 'rs.r1', 'FailRange', cur) cur.execute("rollback to savepoint x;") # clear the adapters to allow precise count by scripts/refcounter.py for r in [ra1, ra2, rars2, rars3]: del ext.adapters[r.range, ext.ISQLQuote]
def test_register_range_adapter(self): from psycopg2.extras import Range, register_range cur = self.conn.cursor() cur.execute("create type textrange as range (subtype=text)") rc = register_range('textrange', 'TextRange', cur) TextRange = rc.range self.assert_(issubclass(TextRange, Range)) self.assertEqual(TextRange.__name__, 'TextRange') r = TextRange('a', 'b', '(]') cur.execute("select %s", (r,)) r1 = cur.fetchone()[0] self.assertEqual(r1.lower, 'a') self.assertEqual(r1.upper, 'b') self.assert_(not r1.lower_inc) self.assert_(r1.upper_inc) cur.execute("select %s", ([r, r, r],)) rs = cur.fetchone()[0] self.assertEqual(len(rs), 3) for r1 in rs: self.assertEqual(r1.lower, 'a') self.assertEqual(r1.upper, 'b') self.assert_(not r1.lower_inc) self.assert_(r1.upper_inc) # clear the adapters to allow precise count by scripts/refcounter.py del ext.adapters[rc.range, ext.ISQLQuote]
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.assertQuotedEqual( curs.mogrify("%s", (obj,)), b"""'{"a": 123}'""") finally: del psycopg2.extensions.adapters[dict, ext.ISQLQuote]
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_register_range_adapter(self): from psycopg2.extras import Range, register_range cur = self.conn.cursor() cur.execute("create type textrange as range (subtype=text)") rc = register_range('textrange', 'TextRange', cur) TextRange = rc.range self.assert_(issubclass(TextRange, Range)) self.assertEqual(TextRange.__name__, 'TextRange') r = TextRange('a', 'b', '(]') cur.execute("select %s", (r,)) r1 = cur.fetchone()[0] self.assertEqual(r1.lower, 'a') self.assertEqual(r1.upper, 'b') self.assert_(not r1.lower_inc) self.assert_(r1.upper_inc) cur.execute("select %s", ([r,r,r],)) rs = cur.fetchone()[0] self.assertEqual(len(rs), 3) for r1 in rs: self.assertEqual(r1.lower, 'a') self.assertEqual(r1.upper, 'b') self.assert_(not r1.lower_inc) self.assert_(r1.upper_inc) # clear the adapters to allow precise count by scripts/refcounter.py del ext.adapters[rc.range, ext.ISQLQuote]
