如何通过FastAPI触发运行长耗时大型Python脚本并实现后台执行？

方案选择说明 #

FastAPI 自带的 BackgroundTasks 仅适用于短周期、丢失后无影响的轻量任务，1-2周的长周期任务会面临服务重启丢失、进程崩溃无重试、无状态追踪等问题，完全不适用该场景。
生产环境最优方案采用 分布式任务队列+进程守护 组合实现，可覆盖任务调度、状态记录、失败重试、异常自动恢复的需求。

完整实现步骤 #

1. 核心依赖选择 #

• 分布式任务队列：Celery，负责任务提交、状态存储、重试逻辑
• 消息中间件：Redis/RabbitMQ，用作Celery的任务队列和结果存储
• 进程守护：systemd/supervisor，保证任务执行进程崩溃后自动拉起

2. 代码实现 #

首先安装所需依赖：

pip install fastapi celery redis uvicorn

Celery 任务配置（celery_app.py）

from celery import Celery

# 初始化Celery实例，broker和backend均使用Redis
celery = Celery(
    "long_running_tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)

# 定义长周期任务，最多重试3次
@celery.task(bind=True, max_retries=3)
def exec_long_script(self):
    import subprocess
    # 调用你的长周期脚本，输出重定向到日志文件方便排查
    log_file = open("/data/long_script.log", "a", encoding="utf-8")
    result = subprocess.run(
        ["bash", "/path/to/your/target_script.sh"],
        stdout=log_file,
        stderr=log_file,
        text=True
    )
    return {"return_code": result.returncode}

FastAPI 接口实现（main.py）

from fastapi import FastAPI
from celery_app import exec_long_script

app = FastAPI()

@app.get("/")
def start_process():
    # 异步提交任务到Celery队列，立即返回无需等待执行
    task = exec_long_script.delay()
    return {
        "message": "Process has been started successfully",
        "task_id": task.id # 后续可通过该ID查询任务状态
    }

# 可选：任务状态查询接口
@app.get("/task/{task_id}")
def get_task_status(task_id: str):
    from celery_app import celery
    task_result = celery.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": task_result.state,
        "result": task_result.result if task_result.state == "SUCCESS" else None,
        "error": str(task_result.info) if task_result.state == "FAILURE" else None
    }

3. 部署配置 #

1. 启动Redis服务，确保Celery可以正常连接
2. 启动Celery Worker进程：

   celery -A celery_app worker --loglevel=info --concurrency=1
   

   其中concurrency参数为同时执行的任务数，长周期任务建议设置为1或较小值
3. 配置systemd进程守护，保证Worker崩溃后自动重启，示例配置/etc/systemd/system/celery-worker.service：

   [Unit]
   Description=Celery Long Task Worker
   After=network.target redis.target
   
   [Service]
   User=root
   WorkingDirectory=/path/to/your/project/directory
   ExecStart=/usr/bin/python3 -m celery -A celery_app worker --loglevel=info --concurrency=1
   Restart=always
   RestartSec=10
   
   [Install]
   WantedBy=multi-user.target
   

   配置完成后执行以下命令生效：

   systemctl daemon-reload
   systemctl enable --now celery-worker
   

4. 启动FastAPI服务即可正常使用

极简临时方案（仅适用于测试） #

如果是临时测试场景，不需要状态追踪、重试等能力，也可以直接用subprocess.Popen启动脱离主进程的后台子进程，实现代码如下：

import subprocess
import os
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def start_process():
    log_file = open("/data/long_script.log", "a")
    # preexec_fn=os.setsid 让子进程脱离父进程生命周期，主进程退出不会杀掉子进程
    subprocess.Popen(
        ["bash", "/path/to/your/target_script.sh"],
        stdout=log_file,
        stderr=log_file,
        preexec_fn=os.setsid
    )
    return {"message": "Process has been started successfully"}

该方案的缺点是进程崩溃后不会自动恢复，服务重启后无法追踪之前运行的任务状态，生产环境不推荐使用。

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