Flask路由中后台运行外部Python代码，解决请求超时问题

遇到这种网关超时的问题太常见了——Flask默认是同步处理请求，长时间运行的函数会把请求线程占住，直到超时。下面给你几个靠谱的解决办法，按复杂度和适用场景分：

方案一：用后台线程（快速实现，适合小型项目） #

如果你的项目规模不大，不需要复杂的任务管理，直接用Python的threading模块把耗时任务丢到后台线程里就行。注意要处理好Flask的上下文，还有避免线程安全问题。

修改你的路由代码：

import threading
from flask import current_app

@app.route('/result/<case_id>')
def result(case_id):
    # 定义后台任务函数，要把需要的参数传进去
    def background_task():
        # 如果任务里需要用到Flask的配置（比如UPLOAD_FOLDER），要手动推送上下文
        with current_app.app_context():
            USER_FOLDER = os.path.join(UPLOAD_FOLDER, case_id)
            Analysis_code.main(USER_FOLDER, case_id, 
                              f"{case_id}_mRNA_file.txt", 
                              f"{case_id}_lncRNA_file.txt", 
                              f"{case_id}_miRNA_file.txt")
    
    # 启动后台线程
    threading.Thread(target=background_task, daemon=True).start()
    
    # 立刻返回页面，不用等任务完成
    return render_template('test.html', case_id=case_id)

注意：后台线程设置daemon=True，这样当Flask进程退出时，线程会自动终止。如果任务必须执行完，可以去掉这个参数，但要注意进程意外退出的情况。另外，如果任务涉及数据库操作，要确保线程安全。

方案二：用Celery（适合中大型项目，支持任务监控/重试） #

如果你的项目需要更完善的任务管理——比如任务队列、失败重试、进度监控，Celery是标准解决方案。需要配合消息队列（比如Redis或者RabbitMQ）使用。

步骤1：安装依赖 #

pip install celery redis

步骤2：配置Celery #

在你的Flask项目根目录创建celery_config.py：

import os
from celery import Celery
from flask import Flask

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)
    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)
    celery.Task = ContextTask
    return celery

# 初始化Flask app并配置Celery
app = Flask(__name__)
app.config.update(
    CELERY_BROKER_URL='redis://localhost:6379/0',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)

celery = make_celery(app)

步骤3：把耗时任务改成Celery任务 #

在Analysis_code.py或者单独的任务文件里定义：

from celery_config import celery
# 导入你的correlation和bipartite_network函数

@celery.task
def run_analysis(user_folder, case_id, mrna_file, lncrna_file, mirna_file):
    correlation(user_folder, case_id, mrna_file, lncrna_file, mirna_file)
    bipartite_network(user_folder, case_id)

步骤4：修改Flask路由 #

from celery_config import celery
from your_tasks_file import run_analysis

@app.route('/result/<case_id>')
def result(case_id):
    USER_FOLDER = os.path.join(UPLOAD_FOLDER, case_id)
    # 发送任务到Celery队列，立刻返回
    run_analysis.delay(USER_FOLDER, case_id, 
                      f"{case_id}_mRNA_file.txt", 
                      f"{case_id}_lncRNA_file.txt", 
                      f"{case_id}_miRNA_file.txt")
    return render_template('test.html', case_id=case_id)

步骤5：启动Celery worker #

在终端运行：

celery -A celery_config worker --loglevel=info

这样任务就会在Celery worker里后台运行，Flask请求不会被阻塞，还能通过Celery的API查看任务状态、结果。

方案三：用Flask-APScheduler（适合定时/一次性任务） #

如果只是需要触发一次性的后台任务，Flask-APScheduler也可以用，它基于APScheduler，支持多种触发器。

步骤1：安装 #

pip install flask-apscheduler

步骤2：配置和使用 #

from flask import Flask
from flask_apscheduler import APScheduler

app = Flask(__name__)
app.config['SCHEDULER_API_ENABLED'] = True
scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()

def background_task(user_folder, case_id, mrna_file, lncrna_file, mirna_file):
    Analysis_code.main(user_folder, case_id, mrna_file, lncrna_file, mirna_file)

@app.route('/result/<case_id>')
def result(case_id):
    USER_FOLDER = os.path.join(UPLOAD_FOLDER, case_id)
    # 添加一次性任务，立刻执行
    scheduler.add_job(
        id=f"analysis_{case_id}",
        func=background_task,
        args=[USER_FOLDER, case_id, 
              f"{case_id}_mRNA_file.txt", 
              f"{case_id}_lncRNA_file.txt", 
              f"{case_id}_miRNA_file.txt"],
        trigger='date'
    )
    return render_template('test.html', case_id=case_id)

选择建议 #

• 如果是小项目/快速验证，优先用后台线程，零额外依赖，实现简单。
• 如果需要任务管理、重试、监控，选Celery，工业级方案，扩展性强。
• 如果需要定时任务或者简单的一次性后台任务，Flask-APScheduler也很方便。

内容的提问来源于stack exchange，提问作者st504132005