VS Code 联动调试 Flask 与 Celery -Yòuyou

本文讲如何在 VS Code 中调试 Flask 接口和 Celery 后台任务：给代码设置断点，发送一次请求，让程序停下来，查看变量，再逐行执行。

示例只做一次加法：接口收到 2 和 3，把加法交给后台任务，最后得到 5。调试时会停两次：

在 Flask 接口中暂停，检查收到的两个数；
在 Celery 任务中暂停，检查计算过程和结果。

1. 为什么要启动两个调试程序

Flask 负责接收请求。Celery 用来安排后台任务，执行任务的程序叫 Worker。它们是两个分别运行的 Python 进程。

例如，接口收到“生成报表”的请求后，可以先把任务交出去，让 Worker 生成报表，接口不必一直等到报表完成。本例把生成报表换成加法，方便观察代码执行过程。

Flask 和 Worker 之间通过 Redis 传递任务：

请求：计算 2 + 3
    ↓
Flask 接口：发送加法任务
    ↓
Redis：暂存任务消息
    ↓
Celery Worker：取出任务，执行加法，得到 5

因此，只启动 Flask 的调试器，能停在接口代码里，却看不到 Worker 执行加法的过程。VS Code 需要分别启动 Flask 和 Worker 的调试器。

后面会把这两个启动配置组成一组，按一次 F5 就一起启动。这就是本文的“联动调试”。Redis 只负责传递消息，在调试前启动即可。

2. 准备能运行的示例

下面的命令适用于 macOS / Linux。Windows 可以在 WSL 中执行，并用 VS Code 打开对应的 WSL 文件夹。

环境需要 Python 3.11 或以上版本、VS Code 和运行中的 Docker。示例目录如下：

flask-celery-debug/
├── .vscode/
│   └── launch.json
├── tasks.py
└── web.py

安装 Python 依赖

创建 flask-celery-debug 文件夹，在该目录的终端中执行：

python3 -m venv .venv
.venv/bin/python -m pip install 'Flask>=3.1,<3.2' 'celery[redis]>=5.6,<5.7'

.venv 是这个示例的 Python 环境。后面 VS Code 也要选择其中的 Python，否则调试时可能找不到刚安装的 Flask 或 Celery。

启动 Redis

在终端执行：

docker run -d --rm --name flask-celery-debug-redis \
  -p 127.0.0.1:56379:6379 redis:7-alpine \
  redis-server --save '' --appendonly no
 
docker exec flask-celery-debug-redis redis-cli ping

第二条命令返回 PONG，表示容器内的 Redis 可以响应命令。示例代码通过本机 56379 端口连接它。

定义加法任务

创建 tasks.py：

from celery import Celery
 
 
celery_app = Celery(
    "debug_example",
    broker="redis://127.0.0.1:56379/0",
)
celery_app.conf.task_default_queue = "debug_example"
 
 
@celery_app.task(name="debug_example.add")
def add(a: int, b: int) -> int:
    total = a + b  # Worker 断点放在这一行。
    return total

celery_app 保存 Celery 的配置。这里的 broker 指向负责传递任务消息的 Redis，task_default_queue 指定任务使用的队列。

@celery_app.task 把 add 登记成 Celery 任务。Worker 启动后就能找到这个函数，并执行收到的加法任务。这个文件用于定义任务，运行 Worker 的方式写在后面的调试配置中。

定义提交任务的接口

创建 web.py：

from flask import Flask, jsonify, request
 
from tasks import add
 
 
app = Flask(__name__)
 
 
@app.post("/add")
def submit_add():
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        return jsonify(error="请求体必须是 JSON 对象"), 400
 
    a, b = body.get("a"), body.get("b")
    if type(a) is not int or type(b) is not int:
        return jsonify(error="a 和 b 必须是整数"), 400
 
    job = add.delay(a, b)  # Flask 断点放在这一行。
    return jsonify(task_id=job.id), 202

向 POST /add 发送 {"a": 2, "b": 3} 后，接口先检查参数，再调用 add.delay(a, b)。

delay() 的作用是发送任务。它不会在 Flask 中执行 a + b；加法由 Worker 执行。接口返回的 task_id 是这次任务的编号，不是计算结果。本例在 Worker 日志里查看结果 5。

3. 配置 VS Code 的启动方式

打开文件夹，选择 Python

在 VS Code 中选择 File → Open Folder…（文件 → 打开文件夹），打开 flask-celery-debug。左侧文件列表中应能看到 web.py 和 tasks.py。
在扩展面板安装 Microsoft 的 Python，并确认 Python Debugger 扩展已启用。
按 F1 打开命令面板，搜索 Python: Select Interpreter，选择这个目录下的 .venv/bin/python。如果列表里没有，就用 Enter interpreter path… 指定该文件。

这里选中的 Python 会用于启动后面的两个调试程序。相关设置见 VS Code Python 调试文档。

写入 launch.json

在示例目录下新建 .vscode 文件夹，在里面创建 launch.json。这个文件告诉 VS Code：“按 F5 时启动什么程序，带哪些参数。”

{
  "version": "0.2.0",
  "configurations": [
    {
      "name": "Flask API",
      "type": "debugpy",
      "request": "launch",
      "module": "flask",
      "cwd": "${workspaceFolder}",
      "args": [
        "--app",
        "web:app",
        "run",
        "--debug",
        "--no-debugger",
        "--no-reload",
        "--host",
        "127.0.0.1",
        "--port",
        "5055"
      ],
      "justMyCode": true,
      "console": "integratedTerminal"
    },
    {
      "name": "Celery Worker",
      "type": "debugpy",
      "request": "launch",
      "module": "celery",
      "cwd": "${workspaceFolder}",
      "args": ["-A", "tasks:celery_app", "worker", "--loglevel=INFO", "--pool=solo"],
      "justMyCode": true,
      "console": "integratedTerminal"
    }
  ],
  "compounds": [
    {
      "name": "Flask + Celery",
      "configurations": ["Flask API", "Celery Worker"],
      "stopAll": true
    }
  ]
}

配置分成三部分：

名称	按 F5 后做什么
`Flask API`	启动 Flask，调试 `web.py` 中的接口
`Celery Worker`	启动 Worker，调试 `tasks.py` 中的任务
`Flask + Celery`	一起启动上面两个调试程序

configurations 保存单个程序的启动方式，compounds 把已有配置按名称组合起来。因此，组合中的名称要和前面两个 name 一致。

与示例文件对应的两个入口是：

--app web:app：加载 web.py 中的 Flask 对象 app；
-A tasks:celery_app：加载 tasks.py 中的 Celery 对象 celery_app。

cwd 表示运行目录。${workspaceFolder} 就是刚才打开的 flask-celery-debug 文件夹，Python 从这里查找 web 和 tasks。

4. 放置断点，再按 F5

断点的作用是让程序在执行某一行之前暂停。暂停后，可以查看变量，也可以让程序一行一行往下执行。

打开 web.py，在这一行的行号左侧单击，出现红点：

job = add.delay(a, b)

再打开 tasks.py，在这一行设置第二个断点：

total = a + b

随后打开左侧 Run and Debug（运行和调试） 面板，在顶部下拉框中选择 Flask + Celery，按 F5。

VS Code 会启动两个调试程序。终端中分别出现 Flask 的监听地址 http://127.0.0.1:5055 和 Celery 的 ready 日志后，就可以发请求了。

按 F5 只是启动服务，不会自动调用接口或执行加法。 如果两个断点都还没停下来，这是正常的，程序正在等待请求。

5. 发送请求，查看 Flask 中的参数

用 Terminal → New Terminal（终端 → 新建终端） 打开一个空闲终端，执行：

curl -i -X POST http://127.0.0.1:5055/add \
  -H 'Content-Type: application/json' \
  -d '{"a": 2, "b": 3}'

这条命令向 Flask 发送两个整数。VS Code 随后会停在 web.py 的断点行，并高亮当前行。

此时：

a 的值是 2，b 的值是 3；
add.delay(a, b) 还没有执行，任务尚未发送；
终端里的 curl 还在等接口响应。

变量可以在左侧 VARIABLES（变量） 面板查看，也可以把鼠标移到代码中的 a 或 b 上查看值。

点击调试工具栏的 Continue（继续）。Flask 会执行 delay()，发送任务，然后返回响应。

6. 切到 Worker，单步执行加法

Worker 收到任务后，VS Code 会停在 tasks.py 的：

total = a + b

如果界面没有切到任务代码，在左侧 CALL STACK（调用堆栈） 中找到 Celery Worker，选择其中的 add。变量面板显示的是当前选中位置的变量；查看任务参数时，要选中 Worker 中的 add，而不是 Flask 中的接口函数。

这里也能看到 a = 2、b = 3。因为加法所在行还没执行，此时没有 total 的值。

点击工具栏的 Step Over（单步跳过，F10），执行这一行。随后在变量面板中可以看到：

a = 2
b = 3
total = 5

再点击 Continue（继续），让任务返回。Worker 终端中会出现包含 succeeded 和结果 5 的日志。

curl 收到的则是 HTTP 202 和一个任务编号，例如：

{ "task_id": "本次任务的编号" }

Flask 发送任务后就可以返回，所以 curl 的响应可能在 Worker 暂停期间就已出现，不需要等加法完成。

两个断点之间的过程如下：

Flask 断点：查看 a=2、b=3
    ↓ 点击“继续”
发送任务到 Redis
    ↓ Worker 收到任务
Worker 断点：查看 a=2、b=3
    ↓ 点击“单步跳过”
执行加法，得到 total=5
    ↓ 点击“继续”
任务返回，Worker 日志显示结果 5

对 delay() 点击“单步进入”，不会进入 Worker 中的 add。这两个函数在不同进程里运行，需要在各自的调试程序中设置断点。

7. 修改代码后，停止并重新启动

调试工具栏的红色方块是 Stop（停止）。本例配置了 stopAll: true，手动停止组合中的一个调试程序时，另一个也会停止。

修改 web.py 或 tasks.py 后，停止调试，再选择 Flask + Celery 按 F5。两个程序会重新加载代码。

配置中有两处与这个操作有关：

Flask 的 --no-reload 关闭自动重启；代码修改后由手动重启加载。--no-debugger 关闭 Flask 自带的浏览器调试器，断点由 VS Code 控制。参见 Flask 的 External Debuggers。
Celery 的 --pool=solo 让任务在 Worker 主线程中逐个执行，方便本地设置断点。暂停任务时，这个 Worker 也无法执行下一条任务。这里不讨论生产环境的并发配置。参见 Celery 的 Concurrency。

Redis 是从终端启动的，不会随调试器停止。示例结束后执行：

docker stop flask-celery-debug-redis

该容器使用了 --rm，停止后会删除容器；本例没有保存 Redis 数据。

tags	practice, flask, celery
aliases	—

Yòuyou

探索

VS Code 联动调试 Flask 与 Celery