Function go

lib/sqlalchemy/dialects/mysql/provision.py:139–151

()

def _allow_stale_update_impl(cfg):
    @contextlib.contextmanager
    def go():
        @event.listens_for(cfg.db, "engine_connect")
        def turn_off_snapshot_isolation(conn):
            conn.exec_driver_sql("SET innodb_snapshot_isolation = 'OFF'")
            conn.rollback()

        try:
            yield
        finally:
            event.remove(cfg.db, "engine_connect", turn_off_snapshot_isolation)

            # dispose the pool; quick way to just have those reset
            cfg.db.dispose()

    return go()