Python进阶与并发编程12：把报价协程接到HTTP入口，资源该活多久

前几课的程序有一个明确的主人：asyncio.run(main())启动事件循环，main等待子任务，最后关闭客户端。现在业务同事希望从网页提交“买两把键盘，三家谁最便宜”。如果简单把main()粘到路由函数里，每来一次请求就创建线程池、事件循环或 HTTP 客户端，单次演示仍可能成功，但服务的资源关系已经变了。Web 服务有很多次请求，应用本身只启动一次；我们需要先分清这两种寿命，再把已经验证的业务规则接到入口。

本课结束的代码在checkpoints/12/。它只有健康检查和单项报价接口，没有第13课的全局准入与总预算，没有第14课的重试，更没有批次接口。这里保留每次 HTTP 操作的超时来避免无限等待，但不能据此声称“整个请求只会运行0.8秒”。这个缺口会成为下一课的实际问题。

先兑现E11：取消不是发出一句命令就算结束

E11要求六个慢任务共用Semaphore(2)，取消批次后把它等到退出，确认活动计数归零，再复用同一个信号量完成下一批，同时恢复 ContextVar。完整答案是lessons/12/answer_previous.py，从课程根目录运行：

python lessons/12/answer_previous.py

答案用two_entered事件确认恰有两个任务进入资源区，再调用batch.cancel()。这样取消测试没有依赖“睡0.1秒应该已经开始”的猜测。cancel()只请求取消；真正让 TaskGroup 取消其余孩子、等待孩子执行finally的，是随后那次await batch。调用方捕获CancelledError，是为了在验证取消之后继续测试下一批，并不是业务协程吞掉取消。

active -= 1放在资源区内部的finally，因此任务在睡眠时收到取消，计数也会减少；async with gate.slot()则负责归还槽位。ContextVar 使用token = set(...)与reset(token)配对，恢复的是进入前的值，而非盲目改成某个字符串。父任务的上下文与创建时复制给子任务的上下文也不能混为一谈，所以答案既在批次的 finally 内断言恢复，又在调用方断言仍为预设的caller-before。下一批输入[3,4,5]，在宽容1秒期限内返回[6,8,10]且每行ID为next，峰值不超过2，给出了“资源能再次使用”的行为证据，比读取信号量私有字段更有价值。

这个收尾关系将在 Web 中继续存在：应用 lifespan 等待共享客户端关闭，请求任务的 TaskGroup 等待供应商任务结束。所谓资源管理，不是每层都加一个 try，而是能回答每个资源由谁创建、谁借用、谁最后关闭。

只检查下一批最终完成，尚不能证明两个槽位都回来了：如果意外漏还一个槽位，小批次仍可能串行跑完。答案还清空旧事件，让两个新的holder在同一gate中等待release，确认两者同时进入、active再次达到2，再放行。它不读取Semaphore私有字段，也不复用取消前的历史peak当作恢复证据。

请求合同先于路由代码

我们把调用方可以决定的内容限定为 SKU、数量和教学故障场景。例如：

{"sku":"KB-001","quantity":2,"scenario":"normal"}

SKU 默认KB-001；数量默认2，允许1–100的整数；场景只允许 normal、slow、broken、malformed、flaky。调用方不能提供供应商 URL，因此不能借我们的服务任意访问本机或其他地址。教学场景开关用于稳定复现故障，不应直接照搬成线上客户的业务参数。

数量使用strict=True，因为 JSON 的true、字符串"2"与整数2含义不同。Python 中布尔值是整数的子类，是本系列第一课谈过的类型边界；这里不能只写一个宽松的大小比较便放心。extra="forbid"使拼错的字段和额外 URL 得到422，避免“客户端以为参数生效，服务却忽略了它”。SKU长度有上限，空字符串无意义；未知但格式正确的SKU则仍可向供应商查询，由供应商404给出技术失败结果。

FastAPI根据QuoteRequest解析请求，校验不通过时不会执行报价路由。这个顺序很重要：quantity=0不只是返回错误，还应完全不触碰供应商。下一课的E12答案会验证这一点。响应也声明response_model=QuoteResult，让文档和代码共享结构约束。响应校验不是装饰：如果我们不小心返回不符合约定的数据，服务应暴露程序错误，而不是把错误结构当正常合同发送给客户。

同时信任两边输入，会把错误变成低价

外部调用方是一处边界，供应商 HTTP 响应是另一处。供应商即使返回200，也可能给出unit_cents="unknown"，或在 atlas 路径返回 boreal 的数据。SupplierQuote校验金额和库存都是非负整数；随后_fetch还比较响应中的 supplier、sku与本次请求是否一致。若只校验字段存在，串错货号的合法 JSON 就可能参与排名。

金额全程按“分”存储，合计严格是unit_cents * quantity + shipping_cents。运输费不是每件一份，不能把(unit_cents + shipping_cents) * quantity写成看似自然的公式。正常数量2时，atlas总额26400分，boreal25900分；cedar库存只有1，属于有效但不可售的报价，所以过滤掉而不记供应商失败。failures记录传输、状态码、畸形数据等技术问题；缺货不等于供应商坏了。

成功结果按(total_cents, supplier)排序。虽然三个任务的完成顺序不稳定，HTTP响应的展示顺序是稳定的，最低价同价时也可复现。best是排序后的第一项，若没有可售报价则为null。因此“全部供应商正常但都缺货”与“全部技术失败”都可能 best=null，客户仍可通过failures了解两者差别。

响应还有JSON之前的解码边界：若供应商声明Content-Encoding: gzip，正文却无法解压，HTTPX会在client.get()读取期间抛DecodingError。它继承RequestError而非TransportError，现有传输错误分支无法顺便兜住。我们在HTTP读取边界单独把它转为invalid_data，保留其他供应商结果；稍后的JSON解析和字段校验则分别负责后续边界。

一个应用借用一个客户端

create_app构造 FastAPI 应用；它接收的可选transport用于测试替换网络边界，默认值不设置MockTransport，实际启动就会走真实HTTP。应用模块最后的app = create_app()只构造应用，不在导入时打开网络连接。导入期创建长寿命异步资源，容易把资源绑定到不合适的事件循环，也让测试与多进程启动变得含糊。

lifespan是异步上下文管理器。进入时创建httpx.AsyncClient，把持有它的QuoteService放到app.state；yield之后是应用运行期；退出时由async with等待客户端关闭。每次路由拿到的都是同一服务实例，每次_fetch借用同一连接池。请求结束不关闭客户端，因为别的请求还可能使用它；应用结束才承担最终关闭责任。

这里的共享不是靠模块变量碰巧指向同一个对象，而是由应用生命周期明确拥有。trust_env=False避免机器上的代理环境变量意外接管本地实验请求；供应商地址由QUOTEFLOW_PROVIDER_URL配置，默认回环地址9101端口；关闭重定向跟随则防止在未审查的新地址继续取报价。它们让实验路径更可解释，并不替代真实部署的网络访问控制。

同一个路由中的三种错误，处理位置不同

输入非法由框架返回422；单个供应商503被转换为Failure(supplier="boreal", code="http_503")，其余报价保留，整个查询仍可返回200；代码自己抛出的RuntimeError等程序错误则不伪装成供应商失败。第三种应保留异常与服务器日志，让我们修正缺陷。把整个路由包进except Exception: return []会同时隐藏拼错变量、错误字段和网络失败，调用方只会看见一个莫名其妙的空榜单。

TaskGroup适合表达三个子任务都属于当前查询。每个子任务先在自己的HTTP边界把预期故障转成值，这样boreal503不会取消atlas；真正未预期的异常仍由TaskGroup汇总并取消兄弟任务。我们使用task.result()只发生在TaskGroup正常退出之后，届时每个任务都已结束，不会凭空读取一个尚未完成的结果。

不要在async def路由中调用asyncio.run()。ASGI服务器已经管理事件循环；服务方法直接await异步客户端，让出等待网络的时间。也不要误以为“路由是async，所以里面的任何函数都不会阻塞”：若在里面调用time.sleep()或同步HTTP，事件循环仍会被阻塞。前几课学习的调度规则没有因为加上路由装饰器而改变。

从独立阶段安装，再观察真实数据

先在课程根目录的一个终端启动供应商：python tools/provider_lab.py --port 9101。另一个终端进入checkpoints/12，使用Python3.11建立独立环境，避免默认python恰好指向3.9：

py-3.11-m venv.venv.\.venv\Scripts\python.exe-m pip install-e".[test]".\.venv\Scripts\python.exe-m pytest-q.\.venv\Scripts\python.exe-m uvicorn quoteflow.api:app--host 127.0.0.1--port 9100

Linux使用python3.11 -m venv .venv，后续解释器为.venv/bin/python。不同阶段包名都叫quoteflow，必须各用自己的环境，不能在同一个环境同时editable安装四个阶段。

Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9100/api/quotes-ContentType application/json-Body'{"sku":"KB-001","quantity":2,"scenario":"normal"}'

把场景换成broken，应看到boreal技术失败但atlas仍可售；换成malformed，应看到invalid_data。此阶段还没有重试，所以flaky首试会失败。GET /health返回{"status":"ok"}，表示进程能够处理这个简单入口，不表示所有供应商可用。

阶段测试使用MockTransport控制供应商结果，使用ASGITransport直接执行应用。这样可以精确验证HTTP模型、业务结果及lifespan是否关闭客户端，但没有经过TCP、端口绑定或真实连接池。本课真实演示与进程内测试必须分别描述，后面会有独立的双进程黑盒验收。

一个正常响应，能说明哪些事，不能说明哪些事

如果页面展示了boreal最低价，只能说明本次查询中可售且已成功返回的报价经过了正确计算与比较。它不自动证明报价一定最新，也不证明所有市场供应商都已被覆盖。我们把报价时间、币种、有效期等真实业务字段暂时排除在教学合同之外，避免初学者同时承受太多维度；读者必须能指出这个简化，不能把教学字段直接当采购合同使用。

另外，进程内测试里健康供应商常常马上返回，容易让人误以为异步任务总按创建顺序完成。真实服务器故意让cedar比atlas快、atlas比boreal快，让调度与排序的区别可见。输出quotes的第一个元素是总价最小，而不是最先收到响应。响应中的elapsed_ms是整次聚合经历的时间，也不能拿三家延迟相加后解释为它的必然值。

当你关闭服务进程时，服务器会先处理自己的退出流程，然后离开lifespan；客户端的async with负责释放池中连接。若使用强制终止进程，操作系统仍会回收进程资源，但Python finally不一定执行。课程选择正常退出的生命周期测试，是为了验证应用能控制的清理路径，而非声称代码能抵御任意强杀。

阅读代码时可以手动走一次broken输入：Pydantic确认quantity是2，路由借用服务，三个任务共享客户端，boreal变成Failure，cedar因库存不足被过滤，atlas带26400分总价进入quotes与best。这条逐值路径比背“FastAPI高性能、异步高并发”更能说明程序到底做了什么。

最后检查一次响应字段的单位：total_cents、unit_cents、shipping_cents都是分，elapsed_ms是毫秒，quantity是件数。字段名字承担了避免误读的责任，前端展示元时再做格式化。不要把金额为浮点元和整数分混在同一个字段里，也不要把秒数直接写入带ms后缀的字段。接口一旦被其他客户端使用，这些小单位约定就成了需要长期维护的合同。

E12：让无效数量在网络之前停止

提交quantity=0，断言HTTP状态码为422；供应商测试transport计数必须保持0。测试需要显式进入应用lifespan，否则没有模拟完整启动生命周期。再试quantity=true与quantity=“2”，理解严格整数为何值得保留。验收重点不是把报错截图贴上来，而是证明无效输入没有消耗出站资源。

答案将在第13课给出，完整可运行文件位于lessons/13/answer_previous.py。第12课的核心文件全文如下，均来自本课独立阶段，不包含后续重试、全局预算或批次代码。

本课完整代码

lessons/12/answer_previous.py

"""E11 完整自包含答案：复用共享 gate，恢复调用前 ContextVar。"""importasynciofromcontextlibimportasynccontextmanagerfromcontextvarsimportContextVar request_id=ContextVar("request_id",default="-")classResourceGate:def__init__(self,limit):iflimit<1:raiseValueError("limit must be positive")self.semaphore=asyncio.Semaphore(limit)self.active=0self.peak=0self.two_entered=asyncio.Event()@asynccontextmanagerasyncdefslot(self):asyncwithself.semaphore:self.active+=1self.peak=max(self.peak,self.active)ifself.active==2:self.two_entered.set()try:yieldfinally:self.active-=1asyncdefrun_batch(items,gate,batch_id,delay=0.03):previous=request_id.get()token=request_id.set(batch_id)results=[None]*len(items)try:asyncdefone(index,item):asyncwithgate.slot():awaitasyncio.sleep(delay)results[index]={"value":item*2,"request_id":request_id.get()}asyncwithasyncio.TaskGroup()asgroup:forindex,iteminenumerate(items):group.create_task(one(index,item))returnresultsfinally:request_id.reset(token)assertrequest_id.get()==previousasyncdefmain():token=request_id.set("caller-before")try:gate=ResourceGate(2)batch=asyncio.create_task(run_batch(list(range(6)),gate,"cancel-me",delay=5))awaitgate.two_entered.wait()assertgate.active==2batch.cancel()try:awaitbatchexceptasyncio.CancelledError:pass# 调用方确认收到了取消；run_batch 本身没有吞取消。else:raiseAssertionError("外部取消必须传播到调用方")assertgate.active==0# 旧peak可能来自取消前；让两位新holder同时进入，证明两槽都归还。gate.two_entered.clear()release=asyncio.Event()asyncdefhold_slot():asyncwithgate.slot():awaitrelease.wait()asyncwithasyncio.TaskGroup()asgroup:group.create_task(hold_slot())group.create_task(hold_slot())awaitasyncio.wait_for(gate.two_entered.wait(),timeout=1)assertgate.active==2release.set()assertgate.active==0asyncwithasyncio.timeout(1):rows=awaitrun_batch([3,4,5],gate,"next")assert[row["value"]forrowinrows]==[6,8,10]assert{row["request_id"]forrowinrows}=={"next"}assertgate.active==0andgate.peak<=2assertrequest_id.get()=="caller-before"finally:request_id.reset(token)print("E11 passed: active=0, peak<=2, results=[6,8,10], context restored")if__name__=="__main__":asyncio.run(main())

checkpoints/12/pyproject.toml

[build-system] requires = ["setuptools>=68,<81"] build-backend = "setuptools.build_meta" [project] name = "quoteflow-course" version = "1.0.0" description = "Python 3.11 bounded concurrency teaching service" requires-python = ">=3.11" dependencies = ["fastapi==0.115.12", "httpx==0.28.1", "uvicorn==0.34.2", "pydantic>=2.7,<3"] [project.optional-dependencies] test = ["pytest==8.3.5"] [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["src"]

checkpoints/12/src/quoteflow/__init__.py

"""Standalone course checkpoint."""

checkpoints/12/src/quoteflow/models.py

fromtypingimportAnnotated,LiteralfrompydanticimportBaseModel,ConfigDict,Field Supplier=Literal["atlas","boreal","cedar"]Scenario=Literal["normal","slow","broken","malformed","flaky"]NonnegativeInt=Annotated[int,Field(strict=True,ge=0)]classQuoteRequest(BaseModel):"""入口严格校验；布尔值、字符串数量和调用方自定义 URL 均被拒绝。"""model_config=ConfigDict(extra="forbid")sku:str=Field(default="KB-001",min_length=1,max_length=64)quantity:int=Field(default=2,strict=True,ge=1,le=100)scenario:Scenario="normal"classSupplierQuote(BaseModel):"""供应商是另一处不可信边界：金额分与库存必须为非负整数。"""model_config=ConfigDict(extra="forbid",frozen=True)supplier:Supplier sku:strunit_cents:NonnegativeInt shipping_cents:NonnegativeInt stock:NonnegativeIntclassRankedQuote(SupplierQuote):total_cents:NonnegativeIntclassFailure(BaseModel):supplier:Supplier code:strclassQuoteResult(BaseModel):sku:strquantity:intquotes:list[RankedQuote]failures:list[Failure]best:RankedQuote|Noneelapsed_ms:floatrequest_id:str

checkpoints/12/src/quoteflow/service.py

"""First web stage: shared client, no global admission or total deadline yet."""importasynciofromtimeimportperf_counterfromuuidimportuuid4importhttpxfrompydanticimportValidationErrorfrom.modelsimportFailure,QuoteRequest,QuoteResult,RankedQuote,SupplierQuote SUPPLIERS=("atlas","boreal","cedar")classQuoteService:def__init__(self,client:httpx.AsyncClient):self.client=clientasyncdef_fetch(self,supplier,query,request_id):try:response=awaitself.client.get(f"/quotes/{supplier}",params={"sku":query.sku,"scenario":query.scenario},headers={"X-Request-ID":request_id},)excepthttpx.DecodingError:# gzip 等响应体解码在 client.get 内发生，属于供应商数据错误。returnFailure(supplier=supplier,code="invalid_data")excepthttpx.TimeoutException:returnFailure(supplier=supplier,code="timeout")excepthttpx.TransportError:returnFailure(supplier=supplier,code="transport_error")ifresponse.status_code!=200:returnFailure(supplier=supplier,code=f"http_{response.status_code}")try:quote=SupplierQuote.model_validate(response.json())except(ValueError,ValidationError):returnFailure(supplier=supplier,code="invalid_data")ifquote.supplier!=supplierorquote.sku!=query.sku:returnFailure(supplier=supplier,code="invalid_data")returnquoteasyncdefquote(self,query:QuoteRequest)->QuoteResult:started,request_id=perf_counter(),uuid4().hexasyncwithasyncio.TaskGroup()asgroup:tasks=[group.create_task(self._fetch(name,query,request_id))fornameinSUPPLIERS]quotes,failures=[],[]fortaskintasks:value=task.result()ifisinstance(value,Failure):failures.append(value)elifvalue.stock>=query.quantity:quotes.append(RankedQuote(**value.model_dump(),total_cents=(value.unit_cents*query.quantity+value.shipping_cents)))quotes.sort(key=lambdaquote:(quote.total_cents,quote.supplier))returnQuoteResult(sku=query.sku,quantity=query.quantity,quotes=quotes,failures=failures,best=quotes[0]ifquoteselseNone,elapsed_ms=round((perf_counter()-started)*1000,3),request_id=request_id)

checkpoints/12/src/quoteflow/api.py

importosfromcontextlibimportasynccontextmanagerimporthttpxfromfastapiimportFastAPI,Requestfrom.modelsimportQuoteRequest,QuoteResultfrom.serviceimportQuoteServicedefcreate_app(*,transport=None)->FastAPI:@asynccontextmanagerasyncdeflifespan(app):# 启动时创建一次，应用关闭时等待 aclose；每次请求只借用客户端。asyncwithhttpx.AsyncClient(base_url=os.getenv("QUOTEFLOW_PROVIDER_URL","http://127.0.0.1:9101"),timeout=httpx.Timeout(0.8),transport=transport,trust_env=False,follow_redirects=False,)asclient:app.state.service=QuoteService(client)yieldapp=FastAPI(title="QuoteFlow",version="1.0.0",lifespan=lifespan)@app.get("/health")asyncdefhealth():return{"status":"ok"}@app.post("/api/quotes",response_model=QuoteResult)asyncdefquotes(query:QuoteRequest,request:Request):returnawaitrequest.app.state.service.quote(query)returnapp app=create_app()