Flask应用上下文问题解析与解决方案：从错误日志到完美修复

RuntimeError: Working outside of application context.
This typically means that you attempted to use functionality that needed
the current application. To solve this, set up an application context
with app.app_context(). See the documentation for more information.

2025-05-04 22:47:55,208 - INFO - [1] 处理 夜郎全国5-4号 的数据...
2025-05-04 22:47:55,300 - WARNING - 没有查询到匹配的手机号,二次匹配
2025-05-04 22:47:55,402 - INFO - 查询到 0 个匹配的手机号
2025-05-04 22:47:55,413 - ERROR - 处理出错: Working outside of application context.
...
File "D:\桌面\doudian-phone-tool\doudian\deal_excel_file.py", line 263, in save_order_to_db
    db.session.rollback()
RuntimeError: Working outside of application context.

from flask import current_app
from myapp.models import db

def background_task():
    # ❌ 错误：没有应用上下文
    db.session.query(User).all()  # 抛出 RuntimeError

from flask import current_app

def process_single_thread(records, userId):
    with current_app.app_context():  # ✅ 手动创建上下文
        try:
            # 数据库操作
            save_order_to_db(record, userId, status='失败')
        except Exception as e:
            db.session.rollback()
            raise e

from flask import Blueprint, jsonify

bp = Blueprint('orders', __name__)

@bp.route('/process-order', methods=['POST'])
def process_order():
    data = request.get_json()
    process_single_thread(data['records'], data['userId'])  # ✅ 自动有上下文
    return jsonify({"status": "success"})

from flask_executor import Executor

executor = Executor(app)

@bp.route('/start-task', methods=['POST'])
def start_task():
    executor.submit(process_single_thread, records, userId)  # ✅ 自动管理上下文
    return jsonify({"status": "started"})

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app():
    app = Flask(__name__)
    db.init_app(app)  # ✅ 正确初始化
    return app

from flask import current_app

def process_single_thread(records, userId):
    with current_app.app_context():  # ✅ 确保有应用上下文
        try:
            # 处理数据
            matched_phones = query_matching_phones(records)
            if not matched_phones:
                raise ValueError("没有查询到匹配的手机号")

            save_order_to_db(records, userId, status='成功')
        except Exception as e:
            current_app.logger.error(f"处理出错: {str(e)}")
            save_order_to_db(records, userId, status='失败')
            raise

def save_order_to_db(record, userId, status):
    try:
        order = Order(
            user_id=userId,
            data=record,
            status=status
        )
        db.session.add(order)
        db.session.commit()
    except Exception as e:
        db.session.rollback()  # ✅ 现在不会报错
        raise

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app():
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///orders.db'
    db.init_app(app)

    # 注册蓝图
    from .routes import orders_bp
    app.register_blueprint(orders_bp)

    return app