Python 现代开发学习笔记

(十九)把任务接到 FastAPI

1. 当前用户来自依赖

本章 HTTP 示例与前面的 FastAPI 使用同一种分层:路由读参数,身份依赖确认用户,服务函数推进任务,存储器提交计划。

为了直接观察 Alice 与 Bob 两个身份,项目提供两个练习令牌:alice-lesson-token、bob-lesson-token,也可以通过 STUDY_ALICE_TOKEN、STUDY_BOB_TOKEN 环境变量替换。继续接入第十三章的书籍 API 时,将这个映射替换为已经实现的 current_user 依赖,由登录后的用户 id 构造上下文。

认证身份进入 thread_id、检索归属过滤与业务记录,另一用户读取同名任务时得到不同键

内部 thread_id 使用 alice:numpy-1。这个前缀负责区分存储位置;HTTP 身份依赖才负责确认来访者是谁。工具参数里没有 owner,当前用户也不能通过请求体替换成另一个人。

2. 用两次 HTTP 请求完成确认

# 在一个终端启动服务;另一个终端执行下面的客户端。
uv run study-agent --data data serve
# 在另一个终端运行此客户端;首次使用新的任务编号。
import httpx

headers = {"Authorization": "Bearer alice-lesson-token"}
with httpx.Client(base_url="http://127.0.0.1:8783", headers=headers, timeout=30) as client:
    response = client.post("/runs/http-1", json={
        "question": "NumPy 切片怎样避免改动原数组?",
        "write_plan": True,
        "minutes": 35,
    })
    response.raise_for_status()
    print(response.json()["pending"])  # 此时先看将要执行的动作。
    decision = input("确认保存这份计划?输入 y 批准,其余输入拒绝:").strip().lower()
    approved = client.post("/runs/http-1/approval", json={"approved": decision == "y"})
    approved.raise_for_status()
    print(approved.json()["state"]["status"])  # 批准得到 saved,拒绝得到 rejected。
    print(client.get("/plans").json())  # 正式记录来自业务数据库。

NewRun 在调用图之前检查问题长度与时长;确认模型使用 StrictBool,字符串 "true" 不会冒充布尔值。没有令牌返回 401,Bob 读取 Alice 的同名任务返回 404;任务没有等待确认时再次批准返回 409。输入、身份、状态三类检查分别落在相应步骤中。

HTTP 服务默认使用离线回放,与命令行的模型选择分开。扩展本机模型服务时,可以在 lifespan 建立模型与索引,把依赖注入每次运行;需要独立任务队列时,则把 task id、任务状态、取消与结果查询接到第八章及第十三章介绍的后台执行方式。

笔记正文与全部题解 · Ctrl / ⌘ K