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

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

项目：psycopg2-for-aws-lambda 作者：iwitaly | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：pg2kinesis 作者：handshake | 项目源码 | 文件源码

def test_delete_slot(slot):
    with patch.object(psycopg2.ProgrammingError, 'pgcode',
                      new_callable=PropertyMock,
                      return_value=psycopg2.errorcodes.UNDEFINED_OBJECT):
        pe = psycopg2.ProgrammingError()
        slot._repl_cursor.drop_replication_slot = Mock(side_effect=pe)
        slot.delete_slot()
    slot._repl_cursor.drop_replication_slot.assert_called_with('pg2kinesis')

    with patch.object(psycopg2.ProgrammingError, 'pgcode',
                      new_callable=PropertyMock,
                      return_value=-1):
        pe = psycopg2.ProgrammingError()
        slot._repl_cursor.create_replication_slot = Mock(side_effect=pe)
        with pytest.raises(psycopg2.ProgrammingError) as e_info:
            slot.delete_slot()
            slot._repl_cursor.drop_replication_slot.assert_called_with('pg2kinesis')

            assert e_info.value.pgcode == -1

    slot._repl_cursor.create_replication_slot = Mock(side_effect=Exception)
    with pytest.raises(Exception):
        slot.delete_slot()
        slot._repl_cursor.drop_replication_slot.assert_called_with('pg2kinesis')

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：nmbs-realtime-feed 作者：datamindedbe | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：aws-lambda-redshift-copy 作者：christianhxc | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError, e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError, e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：PyEloqua-Examples 作者：colemanja91 | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError as e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError as e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：pg2kinesis 作者：handshake | 项目源码 | 文件源码

def test_create_slot(slot):

    with patch.object(psycopg2.ProgrammingError, 'pgcode',
                      new_callable=PropertyMock,
                      return_value=psycopg2.errorcodes.DUPLICATE_OBJECT):
        pe = psycopg2.ProgrammingError()


        slot._repl_cursor.create_replication_slot = Mock(side_effect=pe)
        slot.create_slot()
        slot._repl_cursor.create_replication_slot.assert_called_with('pg2kinesis',
                                                                     slot_type=psycopg2.extras.REPLICATION_LOGICAL,
                                                                     output_plugin=u'test_decoding')
    with patch.object(psycopg2.ProgrammingError, 'pgcode',
                          new_callable=PropertyMock,
                          return_value=-1):
        pe = psycopg2.ProgrammingError()
        slot._repl_cursor.create_replication_slot = Mock(side_effect=pe)

        with pytest.raises(psycopg2.ProgrammingError) as e_info:
            slot.create_slot()
            slot._repl_cursor.create_replication_slot.assert_called_with('pg2kinesis',
                                                                         slot_type=psycopg2.extras.REPLICATION_LOGICAL,
                                                                         output_plugin=u'test_decoding')
        assert e_info.value.pgcode == -1

        slot._repl_cursor.create_replication_slot = Mock(side_effect=Exception)
    with pytest.raises(Exception):
        slot.create_slot()
        slot._repl_cursor.create_replication_slot.assert_called_with('pg2kinesis',
                                                                         slot_type=psycopg2.extras.REPLICATION_LOGICAL,
                                                                         output_plugin=u'test_decoding')

项目：pg2kinesis 作者：handshake | 项目源码 | 文件源码

def create_slot(self):
        logger.info('Creating slot %s' % self.slot_name)
        try:
            self._repl_cursor.create_replication_slot(self.slot_name,
                                                      slot_type=psycopg2.extras.REPLICATION_LOGICAL,
                                                      output_plugin='test_decoding')
        except psycopg2.ProgrammingError as p:
            # Will be raised if slot exists already.
            if p.pgcode != psycopg2.errorcodes.DUPLICATE_OBJECT:
                logger.error(p)
                raise
            else:
                logger.info('Slot %s is already present.' % self.slot_name)

项目：pg2kinesis 作者：handshake | 项目源码 | 文件源码

def delete_slot(self):
        logger.info('Deleting slot %s' % self.slot_name)
        try:
            self._repl_cursor.drop_replication_slot(self.slot_name)
        except psycopg2.ProgrammingError as p:
            # Will be raised if slot exists already.
            if p.pgcode != psycopg2.errorcodes.UNDEFINED_OBJECT:
                logger.error(p)
                raise
            else:
                logger.info('Slot %s was not found.' % self.slot_name)

项目：psycopg2-for-aws-lambda 作者：iwitaly | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError as e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：psycopg2-for-aws-lambda 作者：iwitaly | 项目源码 | 文件源码

def test_cleanup_on_badconn_close(self):
        # ticket #148
        conn = self.conn
        cur = conn.cursor()
        try:
            cur.execute("select pg_terminate_backend(pg_backend_pid())")
        except psycopg2.OperationalError as e:
            if e.pgcode != psycopg2.errorcodes.ADMIN_SHUTDOWN:
                raise
        except psycopg2.DatabaseError as e:
            # curiously when disconnected in green mode we get a DatabaseError
            # without pgcode.
            if e.pgcode is not None:
                raise

        self.assertEqual(conn.closed, 2)
        conn.close()
        self.assertEqual(conn.closed, 1)

项目：psycopg2-for-aws-lambda 作者：iwitaly | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：ShelbySearch 作者：Agentscreech | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError as e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：userbase-sns-lambda 作者：fartashh | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：Price-Comparator 作者：Thejas-1 | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：nmbs-realtime-feed 作者：datamindedbe | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：aws-lambda-redshift-copy 作者：christianhxc | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：enjoliver 作者：JulienBalestra | 项目源码 | 文件源码

def cockroach_transaction(f):
    def run_transaction(caller):
        while True:
            with MONITOR_COCKROACHDB.observe_transaction(caller):
                try:
                    return f()
                except DatabaseError as e:
                    if not isinstance(e.orig, psycopg2.OperationalError) and \
                            not e.orig.pgcode == psycopg2.errorcodes.SERIALIZATION_FAILURE:
                        raise
                    MONITOR_COCKROACHDB.cockroach_retry_count.labels(caller).inc()

    return run_transaction

项目：PyEloqua-Examples 作者：colemanja91 | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError as e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_

项目：flask 作者：bobohope | 项目源码 | 文件源码

def skip_if_no_superuser(f):
    """Skip a test if the database user running the test is not a superuser"""
    @wraps(f)
    def skip_if_no_superuser_(self):
        from psycopg2 import ProgrammingError
        try:
            return f(self)
        except ProgrammingError, e:
            import psycopg2.errorcodes
            if e.pgcode == psycopg2.errorcodes.INSUFFICIENT_PRIVILEGE:
                self.skipTest("skipped because not superuser")
            else:
                raise

    return skip_if_no_superuser_