Python sqlalchemy.exc 模块，InternalError() 实例源码

def groupfinder(nick, _):
    """
    Finds group for the user id in given request
    :param nick: current user id
    :param request: request
    :return: given group as list or empty list
    """
    # import traceback
    # traceback.print_stack()

    logger('security', 'groupfinder', 'nick: ' + nick)
    try:
        user = DBDiscussionSession.query(User).filter_by(nickname=nick).first()
    except InternalError as i:
        logger('security', 'InternalError', str(i), error=True)
        return []

    if user:
        group = DBDiscussionSession.query(Group).get(user.group_uid)
        if group:
            logger('security', 'groupfinder', 'return [group:' + group.name + ']')
            return ['group:' + group.name]

    logger('security', 'groupfinder', 'return []')
    return []

def includeme(config):
    config.add_request_method(lambda request: {}, 'validated', reify=True)
    config.add_request_method(lambda request: Errors(), 'errors', reify=True)
    config.add_view(view=failed_validation, context=ValidationFailure)
    config.add_view(view=http_error, context=HTTPError)
    config.add_view(view=database_is_read_only, context=InternalError)
    config.add_view(view=refresh_session, context=CSRFTokenError)
    config.add_view(view=refresh_session, context=HTTPForbidden)
    config.add_view(view=refresh_session, context=HTTPPreconditionFailed)
    config.add_view(view=jsondecode_error, context=json.JSONDecodeError)
    config.add_view_predicate('validators', ValidatorsPredicate, weighs_more_than=LAST)

def dbtransaction(app, request, request_ctx, monkeypatch):
    """Temporary DB transaction.

    Use this if you want to operate on the real database but don't want changes to actually affect
    it outside of this test. This works using SQLAlchemy transactions.

    Transactions made outside of the session scope are not rolled back.
    """
    with app.app_context():
        connection = db.engine.connect()
        transaction = connection.begin()

        # Patch Flask-SQLAlchemy to use our connection
        monkeypatch.setattr(db, 'get_engine', lambda *args: connection)

        # Explicitly remove the session so that we'll get a new session every time we go here.
        db.session.remove()

        def teardown():
            # Since we are not committing things to the database directly when
            # testing, initially deferred constraints are not checked. The
            # following statement makes the DB check these constraints. We are
            # executing this command AFTER the tests and NOT BEFORE, because
            # within a transaction the DB is allowed to take temporarily
            # invalid state. Read
            # https://www.postgresql.org/docs/current/static/sql-set-constraints.html
            # for details.

            try:
                connection.execute('SET CONSTRAINTS ALL IMMEDIATE')
            except InternalError:
                # This is the case when we are doing something in the tests
                # that we expect it to fail by executing the statement above.
                # In this case, the transaction will be in an already failed
                # state, executing further SQL statements are ignored and doing
                # so raises an exception.
                pass

            transaction.rollback()
            connection.close()
        request.addfinalizer(teardown)

        return db