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

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因此不能借我们的服务任意访问本机或其他地址。教学场景开关用于稳定复现故障不应直接照搬成线上客户的业务参数。数量使用strictTrue因为 JSON 的true、字符串2与整数2含义不同。Python 中布尔值是整数的子类是本系列第一课谈过的类型边界这里不能只写一个宽松的大小比较便放心。extraforbid使拼错的字段和额外 URL 得到422避免“客户端以为参数生效服务却忽略了它”。SKU长度有上限空字符串无意义未知但格式正确的SKU则仍可向供应商查询由供应商404给出技术失败结果。FastAPI根据QuoteRequest解析请求校验不通过时不会执行报价路由。这个顺序很重要quantity0不只是返回错误还应完全不触碰供应商。下一课的E12答案会验证这一点。响应也声明response_modelQuoteResult让文档和代码共享结构约束。响应校验不是装饰如果我们不小心返回不符合约定的数据服务应暴露程序错误而不是把错误结构当正常合同发送给客户。同时信任两边输入会把错误变成低价外部调用方是一处边界供应商 HTTP 响应是另一处。供应商即使返回200也可能给出unit_centsunknown或在 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。因此“全部供应商正常但都缺货”与“全部技术失败”都可能 bestnull客户仍可通过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.stateyield之后是应用运行期退出时由async with等待客户端关闭。每次路由拿到的都是同一服务实例每次_fetch借用同一连接池。请求结束不关闭客户端因为别的请求还可能使用它应用结束才承担最终关闭责任。这里的共享不是靠模块变量碰巧指向同一个对象而是由应用生命周期明确拥有。trust_envFalse避免机器上的代理环境变量意外接管本地实验请求供应商地址由QUOTEFLOW_PROVIDER_URL配置默认回环地址9101端口关闭重定向跟随则防止在未审查的新地址继续取报价。它们让实验路径更可解释并不替代真实部署的网络访问控制。同一个路由中的三种错误处理位置不同输入非法由框架返回422单个供应商503被转换为Failure(supplierboreal, codehttp_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.9py-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 9100Linux使用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变成Failurecedar因库存不足被过滤atlas带26400分总价进入quotes与best。这条逐值路径比背“FastAPI高性能、异步高并发”更能说明程序到底做了什么。最后检查一次响应字段的单位total_cents、unit_cents、shipping_cents都是分elapsed_ms是毫秒quantity是件数。字段名字承担了避免误读的责任前端展示元时再做格式化。不要把金额为浮点元和整数分混在同一个字段里也不要把秒数直接写入带ms后缀的字段。接口一旦被其他客户端使用这些小单位约定就成了需要长期维护的合同。E12让无效数量在网络之前停止提交quantity0断言HTTP状态码为422供应商测试transport计数必须保持0。测试需要显式进入应用lifespan否则没有模拟完整启动生命周期。再试quantitytrue与quantity“2”理解严格整数为何值得保留。验收重点不是把报错截图贴上来而是证明无效输入没有消耗出站资源。答案将在第13课给出完整可运行文件位于lessons/13/answer_previous.py。第12课的核心文件全文如下均来自本课独立阶段不包含后续重试、全局预算或批次代码。本课完整代码lessons/12/answer_previous.pyE11 完整自包含答案复用共享 gate恢复调用前 ContextVar。importasynciofromcontextlibimportasynccontextmanagerfromcontextvarsimportContextVar request_idContextVar(request_id,default-)classResourceGate:def__init__(self,limit):iflimit1:raiseValueError(limit must be positive)self.semaphoreasyncio.Semaphore(limit)self.active0self.peak0self.two_enteredasyncio.Event()asynccontextmanagerasyncdefslot(self):asyncwithself.semaphore:self.active1self.peakmax(self.peak,self.active)ifself.active2:self.two_entered.set()try:yieldfinally:self.active-1asyncdefrun_batch(items,gate,batch_id,delay0.03):previousrequest_id.get()tokenrequest_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():tokenrequest_id.set(caller-before)try:gateResourceGate(2)batchasyncio.create_task(run_batch(list(range(6)),gate,cancel-me,delay5))awaitgate.two_entered.wait()assertgate.active2batch.cancel()try:awaitbatchexceptasyncio.CancelledError:pass# 调用方确认收到了取消run_batch 本身没有吞取消。else:raiseAssertionError(外部取消必须传播到调用方)assertgate.active0# 旧peak可能来自取消前让两位新holder同时进入证明两槽都归还。gate.two_entered.clear()releaseasyncio.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(),timeout1)assertgate.active2release.set()assertgate.active0asyncwithasyncio.timeout(1):rowsawaitrun_batch([3,4,5],gate,next)assert[row[value]forrowinrows][6,8,10]assert{row[request_id]forrowinrows}{next}assertgate.active0andgate.peak2assertrequest_id.get()caller-beforefinally:request_id.reset(token)print(E11 passed: active0, peak2, results[6,8,10], context restored)if__name____main__:asyncio.run(main())checkpoints/12/pyproject.toml[build-system] requires [setuptools68,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 [fastapi0.115.12, httpx0.28.1, uvicorn0.34.2, pydantic2.7,3] [project.optional-dependencies] test [pytest8.3.5] [tool.pytest.ini_options] testpaths [tests] pythonpath [src]checkpoints/12/src/quoteflow/__init__.pyStandalone course checkpoint.checkpoints/12/src/quoteflow/models.pyfromtypingimportAnnotated,LiteralfrompydanticimportBaseModel,ConfigDict,Field SupplierLiteral[atlas,boreal,cedar]ScenarioLiteral[normal,slow,broken,malformed,flaky]NonnegativeIntAnnotated[int,Field(strictTrue,ge0)]classQuoteRequest(BaseModel):入口严格校验布尔值、字符串数量和调用方自定义 URL 均被拒绝。model_configConfigDict(extraforbid)sku:strField(defaultKB-001,min_length1,max_length64)quantity:intField(default2,strictTrue,ge1,le100)scenario:ScenarionormalclassSupplierQuote(BaseModel):供应商是另一处不可信边界金额分与库存必须为非负整数。model_configConfigDict(extraforbid,frozenTrue)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:strcheckpoints/12/src/quoteflow/service.pyFirst 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.clientclientasyncdef_fetch(self,supplier,query,request_id):try:responseawaitself.client.get(f/quotes/{supplier},params{sku:query.sku,scenario:query.scenario},headers{X-Request-ID:request_id},)excepthttpx.DecodingError:# gzip 等响应体解码在 client.get 内发生属于供应商数据错误。returnFailure(suppliersupplier,codeinvalid_data)excepthttpx.TimeoutException:returnFailure(suppliersupplier,codetimeout)excepthttpx.TransportError:returnFailure(suppliersupplier,codetransport_error)ifresponse.status_code!200:returnFailure(suppliersupplier,codefhttp_{response.status_code})try:quoteSupplierQuote.model_validate(response.json())except(ValueError,ValidationError):returnFailure(suppliersupplier,codeinvalid_data)ifquote.supplier!supplierorquote.sku!query.sku:returnFailure(suppliersupplier,codeinvalid_data)returnquoteasyncdefquote(self,query:QuoteRequest)-QuoteResult:started,request_idperf_counter(),uuid4().hexasyncwithasyncio.TaskGroup()asgroup:tasks[group.create_task(self._fetch(name,query,request_id))fornameinSUPPLIERS]quotes,failures[],[]fortaskintasks:valuetask.result()ifisinstance(value,Failure):failures.append(value)elifvalue.stockquery.quantity:quotes.append(RankedQuote(**value.model_dump(),total_cents(value.unit_cents*query.quantityvalue.shipping_cents)))quotes.sort(keylambdaquote:(quote.total_cents,quote.supplier))returnQuoteResult(skuquery.sku,quantityquery.quantity,quotesquotes,failuresfailures,bestquotes[0]ifquoteselseNone,elapsed_msround((perf_counter()-started)*1000,3),request_idrequest_id)checkpoints/12/src/quoteflow/api.pyimportosfromcontextlibimportasynccontextmanagerimporthttpxfromfastapiimportFastAPI,Requestfrom.modelsimportQuoteRequest,QuoteResultfrom.serviceimportQuoteServicedefcreate_app(*,transportNone)-FastAPI:asynccontextmanagerasyncdeflifespan(app):# 启动时创建一次应用关闭时等待 aclose每次请求只借用客户端。asyncwithhttpx.AsyncClient(base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101),timeouthttpx.Timeout(0.8),transporttransport,trust_envFalse,follow_redirectsFalse,)asclient:app.state.serviceQuoteService(client)yieldappFastAPI(titleQuoteFlow,version1.0.0,lifespanlifespan)app.get(/health)asyncdefhealth():return{status:ok}app.post(/api/quotes,response_modelQuoteResult)asyncdefquotes(query:QuoteRequest,request:Request):returnawaitrequest.app.state.service.quote(query)returnapp appcreate_app()