python类errorcodes()的实例源码

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

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_

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_