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

Python psycopg2.extensions 模块，encodings() 实例源码

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

项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def test_connection_setup(self):
        cur = self.conn.cursor()
        sync_cur = self.sync_conn.cursor()
        del cur, sync_cur

        self.assertTrue(self.conn.async_)
        self.assertTrue(not self.sync_conn.async_)

        # the async connection should be autocommit
        self.assertTrue(self.conn.autocommit)
        self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)

        # check other properties to be found on the connection
        self.assertTrue(self.conn.server_version)
        self.assertTrue(self.conn.protocol_version in (2, 3))
        self.assertTrue(self.conn.encoding in ext.encodings)
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def as_string(self, context):
        # is it a connection or cursor?
        if isinstance(context, ext.connection):
            conn = context
        elif isinstance(context, ext.cursor):
            conn = context.connection
        else:
            raise TypeError("context must be a connection or a cursor")

        a = ext.adapt(self._wrapped)
        if hasattr(a, 'prepare'):
            a.prepare(conn)

        rv = a.getquoted()
        if sys.version_info[0] >= 3 and isinstance(rv, bytes):
            rv = rv.decode(ext.encodings[conn.encoding])

        return rv
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def test_connection_setup(self):
        cur = self.conn.cursor()
        sync_cur = self.sync_conn.cursor()
        del cur, sync_cur

        self.assert_(self.conn.async_)
        self.assert_(not self.sync_conn.async_)

        # the async connection should be autocommit
        self.assert_(self.conn.autocommit)
        self.assertEquals(self.conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)

        # check other properties to be found on the connection
        self.assert_(self.conn.server_version)
        self.assert_(self.conn.protocol_version in (2, 3))
        self.assert_(self.conn.encoding in ext.encodings)
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def as_string(self, context):
        # is it a connection or cursor?
        if isinstance(context, ext.connection):
            conn = context
        elif isinstance(context, ext.cursor):
            conn = context.connection
        else:
            raise TypeError("context must be a connection or a cursor")

        a = ext.adapt(self._wrapped)
        if hasattr(a, 'prepare'):
            a.prepare(conn)

        rv = a.getquoted()
        if sys.version_info[0] >= 3 and isinstance(rv, bytes):
            rv = rv.decode(ext.encodings[conn.encoding])

        return rv
项目：ShelbySearch    作者：Agentscreech    | 项目源码 | 文件源码 
def test_connection_setup(self):
        cur = self.conn.cursor()
        sync_cur = self.sync_conn.cursor()
        del cur, sync_cur

        self.assertTrue(self.conn.async_)
        self.assertTrue(not self.sync_conn.async_)

        # the async connection should be autocommit
        self.assertTrue(self.conn.autocommit)
        self.assertEqual(self.conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)

        # check other properties to be found on the connection
        self.assertTrue(self.conn.server_version)
        self.assertTrue(self.conn.protocol_version in (2, 3))
        self.assertTrue(self.conn.encoding in ext.encodings)
项目：ShelbySearch    作者：Agentscreech    | 项目源码 | 文件源码 
def as_string(self, context):
        # is it a connection or cursor?
        if isinstance(context, ext.connection):
            conn = context
        elif isinstance(context, ext.cursor):
            conn = context.connection
        else:
            raise TypeError("context must be a connection or a cursor")

        a = ext.adapt(self._wrapped)
        if hasattr(a, 'prepare'):
            a.prepare(conn)

        rv = a.getquoted()
        if sys.version_info[0] >= 3 and isinstance(rv, bytes):
            rv = rv.decode(ext.encodings[conn.encoding])

        return rv
项目：nmbs-realtime-feed    作者：datamindedbe    | 项目源码 | 文件源码 
def test_connection_setup(self):
        cur = self.conn.cursor()
        sync_cur = self.sync_conn.cursor()
        del cur, sync_cur

        self.assert_(self.conn.async_)
        self.assert_(not self.sync_conn.async_)

        # the async connection should be autocommit
        self.assert_(self.conn.autocommit)
        self.assertEquals(self.conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)

        # check other properties to be found on the connection
        self.assert_(self.conn.server_version)
        self.assert_(self.conn.protocol_version in (2, 3))
        self.assert_(self.conn.encoding in ext.encodings)
项目：nmbs-realtime-feed    作者：datamindedbe    | 项目源码 | 文件源码 
def as_string(self, context):
        # is it a connection or cursor?
        if isinstance(context, ext.connection):
            conn = context
        elif isinstance(context, ext.cursor):
            conn = context.connection
        else:
            raise TypeError("context must be a connection or a cursor")

        a = ext.adapt(self._wrapped)
        if hasattr(a, 'prepare'):
            a.prepare(conn)

        rv = a.getquoted()
        if sys.version_info[0] >= 3 and isinstance(rv, bytes):
            rv = rv.decode(ext.encodings[conn.encoding])

        return rv
项目：flask    作者：bobohope    | 项目源码 | 文件源码 
def test_connection_setup(self):
        cur = self.conn.cursor()
        sync_cur = self.sync_conn.cursor()
        del cur, sync_cur

        self.assert_(self.conn.async_)
        self.assert_(not self.sync_conn.async_)

        # the async connection should be autocommit
        self.assert_(self.conn.autocommit)
        self.assertEquals(self.conn.isolation_level, ext.ISOLATION_LEVEL_DEFAULT)

        # check other properties to be found on the connection
        self.assert_(self.conn.server_version)
        self.assert_(self.conn.protocol_version in (2, 3))
        self.assert_(self.conn.encoding in ext.encodings)
项目：flask    作者：bobohope    | 项目源码 | 文件源码 
def as_string(self, context):
        # is it a connection or cursor?
        if isinstance(context, ext.connection):
            conn = context
        elif isinstance(context, ext.cursor):
            conn = context.connection
        else:
            raise TypeError("context must be a connection or a cursor")

        a = ext.adapt(self._wrapped)
        if hasattr(a, 'prepare'):
            a.prepare(conn)

        rv = a.getquoted()
        if sys.version_info[0] >= 3 and isinstance(rv, bytes):
            rv = rv.decode(ext.encodings[conn.encoding])

        return rv
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def test_encoding(self):
        conn = self.connect()
        self.assertTrue(conn.encoding in ext.encodings)
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def _logtofile(self, msg, curs):
        msg = self.filter(msg, curs)
        if msg:
            if _sys.version_info[0] >= 3 and isinstance(msg, bytes):
                msg = msg.decode(_ext.encodings[self.encoding], 'replace')
            self._logobj.write(msg + _os.linesep)
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def _logtofile(self, msg, curs):
        msg = self.filter(msg, curs)
        if msg:
            if _sys.version_info[0] >= 3 and isinstance(msg, bytes):
                msg = msg.decode(_ext.encodings[self.encoding], 'replace')
            self._logobj.write(msg + _os.linesep)
项目：psycopg2-for-aws-lambda    作者：iwitaly    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：ShelbySearch    作者：Agentscreech    | 项目源码 | 文件源码 
def test_encoding(self):
        conn = self.connect()
        self.assertTrue(conn.encoding in ext.encodings)
项目：ShelbySearch    作者：Agentscreech    | 项目源码 | 文件源码 
def _logtofile(self, msg, curs):
        msg = self.filter(msg, curs)
        if msg:
            if _sys.version_info[0] >= 3 and isinstance(msg, bytes):
                msg = msg.decode(_ext.encodings[self.encoding], 'replace')
            self._logobj.write(msg + _os.linesep)
项目：ShelbySearch    作者：Agentscreech    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：userbase-sns-lambda    作者：fartashh    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：Price-Comparator    作者：Thejas-1    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：nmbs-realtime-feed    作者：datamindedbe    | 项目源码 | 文件源码 
def test_encoding(self):
        conn = self.connect()
        self.assert_(conn.encoding in ext.encodings)
项目：nmbs-realtime-feed    作者：datamindedbe    | 项目源码 | 文件源码 
def _logtofile(self, msg, curs):
        msg = self.filter(msg, curs)
        if msg:
            if _sys.version_info[0] >= 3 and isinstance(msg, bytes):
                msg = msg.decode(_ext.encodings[self.encoding], 'replace')
            self._logobj.write(msg + _os.linesep)
项目：nmbs-realtime-feed    作者：datamindedbe    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：aws-lambda-redshift-copy    作者：christianhxc    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：PyEloqua-Examples    作者：colemanja91    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)
项目：flask    作者：bobohope    | 项目源码 | 文件源码 
def test_encoding(self):
        conn = self.connect()
        self.assert_(conn.encoding in ext.encodings)
项目：flask    作者：bobohope    | 项目源码 | 文件源码 
def _logtofile(self, msg, curs):
        msg = self.filter(msg, curs)
        if msg:
            if _sys.version_info[0] >= 3 and isinstance(msg, bytes):
                msg = msg.decode(_ext.encodings[self.encoding], 'replace')
            self._logobj.write(msg + _os.linesep)
项目：flask    作者：bobohope    | 项目源码 | 文件源码 
def parse_unicode(self, s, cur):
        """Parse an hstore returning unicode keys and values."""
        if s is None:
            return None

        s = s.decode(_ext.encodings[cur.connection.encoding])
        return self.parse(s, cur)