Python psycopg2.extensions 模块,adapt() 实例源码

我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用psycopg2.extensions.adapt()

项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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]
项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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)
项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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.
项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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]
项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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)
项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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.
项目:ShelbySearch    作者:Agentscreech    | 项目源码 | 文件源码
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]
项目:ShelbySearch    作者:Agentscreech    | 项目源码 | 文件源码
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)
项目:ShelbySearch    作者:Agentscreech    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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)
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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)
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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)
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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)
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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)
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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)
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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())
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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)
项目:userbase-sns-lambda    作者:fartashh    | 项目源码 | 文件源码
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.
项目:Price-Comparator    作者:Thejas-1    | 项目源码 | 文件源码
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())
项目:Price-Comparator    作者:Thejas-1    | 项目源码 | 文件源码
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)
项目:Price-Comparator    作者:Thejas-1    | 项目源码 | 文件源码
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.
项目:nmbs-realtime-feed    作者:datamindedbe    | 项目源码 | 文件源码
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]
项目:nmbs-realtime-feed    作者:datamindedbe    | 项目源码 | 文件源码
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)
项目:nmbs-realtime-feed    作者:datamindedbe    | 项目源码 | 文件源码
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.
项目:aws-lambda-redshift-copy    作者:christianhxc    | 项目源码 | 文件源码
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())
项目:aws-lambda-redshift-copy    作者:christianhxc    | 项目源码 | 文件源码
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)
项目:aws-lambda-redshift-copy    作者:christianhxc    | 项目源码 | 文件源码
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.
项目:PyEloqua-Examples    作者:colemanja91    | 项目源码 | 文件源码
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())
项目:PyEloqua-Examples    作者:colemanja91    | 项目源码 | 文件源码
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)
项目:PyEloqua-Examples    作者:colemanja91    | 项目源码 | 文件源码
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.
项目:flask    作者:bobohope    | 项目源码 | 文件源码
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]
项目:flask    作者:bobohope    | 项目源码 | 文件源码
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)
项目:flask    作者:bobohope    | 项目源码 | 文件源码
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.
项目:jx-sqlite    作者:mozilla    | 项目源码 | 文件源码
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))
项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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())
项目:psycopg2-for-aws-lambda    作者:iwitaly    | 项目源码 | 文件源码
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]