我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用psycopg2.extensions.adapt()。
def test_adapt_most_specific(self): from psycopg2.extensions import adapt, register_adapter, AsIs class A(object): pass class B(A): pass class C(B): pass register_adapter(A, lambda a: AsIs("a")) register_adapter(B, lambda b: AsIs("b")) try: self.assertEqual(b'b', adapt(C()).getquoted()) finally: del psycopg2.extensions.adapters[A, psycopg2.extensions.ISQLQuote] del psycopg2.extensions.adapters[B, psycopg2.extensions.ISQLQuote]
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 getquoted(self): r = self.adapted if r.isempty: return b"'empty'" if not r.lower_inf: # not exactly: we are relying that none of these object is really # quoted (they are numbers). Also, I'm lazy and not preparing the # adapter because I assume encoding doesn't matter for these # objects. lower = adapt(r.lower).getquoted().decode('ascii') else: lower = '' if not r.upper_inf: upper = adapt(r.upper).getquoted().decode('ascii') else: upper = '' return ("'%s%s,%s%s'" % ( r._bounds[0], lower, upper, r._bounds[1])).encode('ascii') # TODO: probably won't work with infs, nans and other tricky cases.
def test_inet_conform(self): from psycopg2.extras import Inet i = Inet("192.168.1.0/24") a = psycopg2.extensions.adapt(i) a.prepare(self.conn) self.assertEqual( filter_scs(self.conn, b("E'192.168.1.0/24'::inet")), a.getquoted()) # adapts ok with unicode too i = Inet(u"192.168.1.0/24") a = psycopg2.extensions.adapt(i) a.prepare(self.conn) self.assertEqual( filter_scs(self.conn, b("E'192.168.1.0/24'::inet")), a.getquoted())
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 getquoted(self): r = self.adapted if r.isempty: return b("'empty'") if not r.lower_inf: # not exactly: we are relying that none of these object is really # quoted (they are numbers). Also, I'm lazy and not preparing the # adapter because I assume encoding doesn't matter for these # objects. lower = adapt(r.lower).getquoted().decode('ascii') else: lower = '' if not r.upper_inf: upper = adapt(r.upper).getquoted().decode('ascii') else: upper = '' return ("'%s%s,%s%s'" % ( r._bounds[0], lower, upper, r._bounds[1])).encode('ascii') # TODO: probably won't work with infs, nans and other tricky cases.
def test_inet_conform(self): from psycopg2.extras import Inet i = Inet("192.168.1.0/24") a = psycopg2.extensions.adapt(i) a.prepare(self.conn) self.assertEqual( filter_scs(self.conn, b"E'192.168.1.0/24'::inet"), a.getquoted()) # adapts ok with unicode too i = Inet("192.168.1.0/24") a = psycopg2.extensions.adapt(i) a.prepare(self.conn) self.assertEqual( filter_scs(self.conn, b"E'192.168.1.0/24'::inet"), a.getquoted())
def quote_value(self, value): if value ==None: return SQL("NULL") if isinstance(value, list): json = convert.value2json(value) return self.quote_value(json) if isinstance(value, basestring) and len(value) > 256: value = value[:256] return SQL(adapt(value))
def test_adapt_subtype(self): from psycopg2.extensions import adapt class Sub(str): pass s1 = "hel'lo" s2 = Sub(s1) self.assertEqual(adapt(s1).getquoted(), adapt(s2).getquoted())
def test_no_mro_no_joy(self): from psycopg2.extensions import adapt, register_adapter, AsIs class A: pass class B(A): pass register_adapter(A, lambda a: AsIs("a")) try: self.assertRaises(psycopg2.ProgrammingError, adapt, B()) finally: del psycopg2.extensions.adapters[A, psycopg2.extensions.ISQLQuote]