企业官网建设流程全解析

Python进阶与并发编程18：把QuoteFlow交给另一个人，才算真正完成项目

独立开发的最后一步，往往不是再加一个接口，而是让一个没有参与开发的人能安装、运行、改动和排查你的服务。若项目只有你当前终端里能启动，或者性能结论只能靠一句“我之前测过”，它还没有形成可交付的工程成果。

这一课把整个专题收束到QuoteFlow。它能查询多家供应商、保留部分成功、按库存与总价选出结果，并对资源容量、截止时间、有限重试和批次顺序作出明确约定。毕业验收不要求把它包装成大而全的电商系统，而是要求你能拿代码和测试解释这些约定，面对故障时知道哪里仍可信、哪里需要进一步观察。

E17完整解答：结论跟着数据走

完整脚本是lessons/18/answer_previous.py，从课程根目录运行：

python lessons/18/answer_previous.py

脚本读取实际报告，找到串行基准，然后用serial_median / current_median计算加速比。它检查请求数量一致、耗时为正、各轮max_active不超配置，最后根据实际样本选择最低整组耗时。并发6的协程如果在你的机器更快，输出会如实变化；它没有把课文表格中的“最佳模式”写死。

E17的三个文字问题也有明确答案。第一，线程延迟从工作线程开始执行算起，async延迟包含Semaphore等待，两个起点不同，不能直接把p95当成同一用户体验指标。第二，12快于6只说明这个12请求的人工等待实验获益，没有覆盖生产配额、内存、数据库和远端饱和，更不能推导1200也合适。第三，spawn启动解释器、导入、参数序列化与结果回传，可能超过小CPU任务本身的计算时间。

如果你的报告缺少某一模式，不要把缺项当作0秒；先检查实验是否完整执行。文件中的status或终端成功行也不能替代具体测量环境。比较两台电脑时，应同时保留报告时间、Python版本、工作量和配置，否则变化可能只是任务变了。

回头看项目，只有三层职责

最终运行代码集中在project/src/quoteflow。models.py定义请求、可信供应商数据和响应结构；service.py管理报价调用、任务、预算与结果组合；api.py负责应用生命周期、路由和HTTP状态映射。文件数量不多，但责任不是按“为了分层而分层”产生的：每一层都有具体边界。

models拒绝布尔数量、负金额和错误字段，让进入排名的数据有可信前提。service不知道怎样画网页，也不把Busy直接写成HTTP响应，它只表达当前容量不足。api把Busy映射到503和Retry-After，并在lifespan中拥有共享客户端。以后即便新增命令行入口，也可以复用service，不需要模拟一个Web请求才能使用业务。

前几课使用dataclass，后面接口模型改用Pydantic，并不表示一种取代了另一种。内存对象实验主要学习对象所有权和规则，HTTP边界则需要结构校验及JSON序列化。我们保持不变的是金额单位、库存过滤、稳定排序与失败语义。连续课程的重点是规则随着实现方式迁移，不能只看文件名是否一成不变。

在一台新电脑上复现

复制整个python-advanced-concurrency-course目录，不能只复制articles。正文里的图片、练习答案、项目和验收工具都是交付的一部分。不要复制旧电脑的虚拟环境；虚拟环境里的解释器路径和平台二进制不适合直接迁移。

Windows PowerShell，在课程根目录执行：

py-3.11-m venv.venv.\.venv\Scripts\python.exe-m pip install-r requirements-verified.txt.\.venv\Scripts\python.exe-m pip install--no-build-isolation-e.\project.\.venv\Scripts\python.exe tools/verify_course.py

这组命令不要求修改PowerShell执行策略，也不要求先激活环境。第一个命令明确选择3.11；后面一直使用刚建立的解释器，避免电脑默认python是3.9或另一版本。依赖文件记录了本次验收用过的实际版本，安装需要可访问Python包源。我们交付的是源码和依赖清单，没有附带离线wheel仓库。

Linux使用python3.11 -m venv .venv建立环境，后续把.\.venv\Scripts\python.exe换为.venv/bin/python即可。Windows spawn路径已经实际运行；本文给出Linux可移植命令，但不把未在Linux系统执行的结果说成Linux验收通过。

两个进程，分别承担什么

自动验收适合回归，学习时也要亲手启动看看。终端A从课程根目录运行：

.\.venv\Scripts\python.exe tools/provider_lab.py--port 9101

终端B仍从课程根目录运行：

.\.venv\Scripts\python.exe-m uvicorn quoteflow.api:app--host 127.0.0.1--port 9100

默认供应商地址就是127.0.0.1:9101。浏览器访问http://127.0.0.1:9100/docs查看OpenAPI，也可以在第三个终端提交：

$body= @{sku='KB-001';quantity=2;scenario='normal'}|ConvertTo-JsonInvoke-RestMethod-Method Post-Uri'http://127.0.0.1:9100/api/quotes'-ContentType'application/json'-Body$body|ConvertTo-Json-Depth 8

应看到boreal总价25900分，atlas26400分，cedar因缺货未入榜。把scenario改成slow，boreal进入failures而atlas仍可用；改成malformed，数据错误不会通过重试“蒙混过去”。两个终端按Ctrl+C停止。它们只绑定回环，目的是学习实验；这不包含公网部署、身份认证或跨机器配额系统。

毕业验收不能只有一排绿色勾

你应能在没有提示答案的情况下完成以下演示，并解释每条结果：

这些检查的执行记录在reports中；脚本会非零退出暴露失败。不要为了演示好看删除失败供应商，也不要把超时预算调到一分钟让所有断言勉强通过。那样改变了合同，验收就失去意义。

毕业改动：用两条测试证明你理解了规则

最终作业不是再复制一次项目。请独立添加两个回归测试，再与文末参考答案比较。

第一条构造同价供应商：atlas和boreal单价100分、运费0、库存2，cedar库存1，购买2件。故意让atlas晚完成。结果仍须是atlas200分、boreal200分，cedar正常过滤，failures为空。这同时检查>=库存边界和稳定平局规则，能抓住把筛选误写成>或按完成顺序返回的缺陷。

第二条让测试适配器抛出TypeError。调用服务必须抛出带TypeError的ExceptionGroup，同时入站计数归零。不能把它写成invalid_data或timeout。我们允许供应商网络失败时部分成功，却不允许自己的程序错误被假装成供应商故障。参考答案在lessons/18/test_graduation.py，运行：

python-m pytest lessons/18/test_graduation.py-q

参考答案不是给测试框架做演示：第一条通过不同等待时间制造到达乱序；第二条检查真正的错误类型以及退出后的资源状态。若你暂时把服务中的异常处理改成except Exception并统一返回Failure，第二条就应该失败。这种能阻止未来退化的断言，才值得长期留下。

面试时怎样介绍这个项目

可以从一个具体问题讲起：“我们要向三家供应商查询报价，一家慢不能拖死整个请求，正常结果还需要统一筛库存和排序。”然后解释三层资源：应用lifespan共享客户端、每进程八个入站槽位、六个出站尝试槽位。最后拿一个真实失败路径说明：slow在0.8秒总预算后返回已完成报价，TaskGroup收尾，本地容量归还，但不声称远端处理立即停止。

被追问为什么用TaskGroup，可以说它把孩子任务的生命周期绑到当前操作，出现未处理错误或取消时收拢任务；不能说它天然支持所有部分成功，部分成功来自我们在供应商边界显式保存可预期Failure。被问为什么不用无限gather，可以说明固定供应商集合与有限批次worker让任务规模有界，Semaphore只控制正在执行的尝试，不能替无界Task列表限制内存。

被问能否部署四个worker，应回答每进程限制会乘上进程数，跨进程配额需要另外设计。被问客户端关闭浏览器是否立刻取消，应区分ASGI断连事件与handler自动取消；本课程证明的是显式Task.cancel路径，没有承诺每个服务器都会自动传播断连。被问生产可用性，要坦诚说明当前是完整教学服务，尚未加入认证、全链路指标、跨进程配额和真实供应商SLA验证。

这些边界不会削弱项目，反而说明你知道代码究竟保护了什么。能够提出下一步需要补哪种证据，比把一个回环实验包装成高并发生产系统更有说服力。

完整参考答案

本课不修改应用模块，因此不再重复整份service.py；它在第15课及最终project中保持一致。下面列出本课两个完整文件：E17报告分析，以及毕业回归测试。专题到此结束，最终作业答案在本篇闭环，不再要求等待不存在的“下一篇”。

lessons/18/answer_previous.py

"""E17：读取实际报告作计算，不写死某台电脑的性能结论。"""importjsonfrompathlibimportPathdefsummarize(path):report=json.loads(path.read_text(encoding="utf-8"))rows=report["rows"]serial=next(rowforrowinrowsifrow["mode"]=="serial")print("measured_at:",report["measured_at"])forrowinrows:assertrow["median_s"]>0peak=max(sample["server_max_active"]forsampleinrow["rounds"])assertpeak<=row["concurrency"]assertrow["requests"]==serial["requests"]speedup=serial["median_s"]/row["median_s"]print(f'{row["mode"]:8}c={row["concurrency"]:2}time={row["median_s"]:.3f}s 'f'speedup={speedup:.2f}x peak={peak}')fastest=min(rows,key=lambdarow:row["median_s"])print("本轮最低整组耗时:",fastest["mode"],fastest["concurrency"])print("仅适用于此组人工I/O负载；不能由此推出生产最优并发或CPU加速结论。")if__name__=="__main__":summarize(Path(__file__).resolve().parents[2]/"reports/benchmark.json")

lessons/18/test_graduation.py

"""毕业改动参考：边界库存、输入合同、程序错误必须可见。"""importasyncioimporthttpximportpytestfromquoteflow.modelsimportQuoteRequestfromquoteflow.serviceimportQuoteServicedeftest_exact_stock_is_sellable_and_price_tie_is_deterministic():asyncdefcase():asyncdefprovider(request):name=request.url.path.rsplit("/",1)[1]# atlas故意最后到，但同价时仍应排在boreal之前。ifname=="atlas":awaitasyncio.sleep(.01)returnhttpx.Response(200,json=dict(supplier=name,sku="KB-001",unit_cents=100,shipping_cents=0,stock=2ifname!="cedar"else1))asyncwithhttpx.AsyncClient(base_url="http://provider",transport=httpx.MockTransport(provider))asclient:result=awaitQuoteService(client).quote(QuoteRequest(quantity=2))assert[(quote.supplier,quote.total_cents)forquoteinresult.quotes]==[("atlas",200),("boreal",200)]assertresult.failures==[]asyncio.run(case())deftest_programming_error_is_not_disguised_as_supplier_failure():asyncdefcase():asyncdefprovider(request):raiseTypeError("deliberate adapter bug")asyncwithhttpx.AsyncClient(base_url="http://provider",transport=httpx.MockTransport(provider))asclient:service=QuoteService(client)withpytest.raises(ExceptionGroup)ascaptured:awaitservice.quote(QuoteRequest())assertany(isinstance(error,TypeError)forerrorincaptured.value.exceptions)assertservice.active_requests==0asyncio.run(case())