Python進(jìn)階與并發(fā)編程14boreal偶爾503重試一次究竟要花掉誰(shuí)的預(yù)算假設(shè)boreal報(bào)價(jià)最低但某次調(diào)用正好遇到服務(wù)切換第一次返回503緊接著就恢復(fù)。第13課會(huì)保留其他供應(yīng)商的結(jié)果因此查詢沒(méi)有完全失敗卻錯(cuò)失了更便宜的報(bào)價(jià)。業(yè)務(wù)希望“再試一下”。這聽(tīng)起來(lái)只需要把HTTP請(qǐng)求放進(jìn)一個(gè)循環(huán)真正危險(xiǎn)的是循環(huán)會(huì)增加出站流量、延長(zhǎng)響應(yīng)時(shí)間并在故障最嚴(yán)重的時(shí)候給供應(yīng)商追加壓力。本課把重試限制為每供應(yīng)商至多兩次嘗試只有503與明確列舉的瞬態(tài)傳輸異常能觸發(fā)第二次。共享出站6、入站8和0.8秒總預(yù)算全部保持。代碼在checkpoints/14仍只有單項(xiàng)接口。請(qǐng)先理解這張預(yù)算賬單再考慮把重試放進(jìn)你自己的服務(wù)。E13答案被拒絕的請(qǐng)求不能偷走一個(gè)槽位完整答案lessons/14/answer_previous.py從第13課階段導(dǎo)入QuoteService運(yùn)行python lessons/14/answer_previous.py。第一條請(qǐng)求的供應(yīng)商處理器設(shè)置entered事件然后等待release。測(cè)試等entered之后再發(fā)第二條確保此時(shí)唯一入站槽位確實(shí)被占用第二條應(yīng)拋Busy而不是悄悄排隊(duì)。放行release之后等待第一條的結(jié)果再檢查active_requests為0最后發(fā)送第三條并成功。這個(gè)順序同時(shí)發(fā)現(xiàn)三種問(wèn)題忙碌時(shí)沒(méi)有拒絕拒絕時(shí)錯(cuò)誤遞增計(jì)數(shù)請(qǐng)求完成時(shí)沒(méi)有在finally中歸還容量。驗(yàn)證槽位復(fù)用是對(duì)用戶可感知行為的斷言不是機(jī)械檢查代碼寫了幾次加減。在HTTP層入口把Busy變?yōu)?03并設(shè)置Retry-After:1。這個(gè)頭表達(dá)客戶端可以稍后重試不構(gòu)成“等待一秒就一定能成功”的保證。自動(dòng)重試行為由客戶端自己決定客戶端如果無(wú)視容量信號(hào)、所有請(qǐng)求立即齊刷刷重發(fā)服務(wù)仍然會(huì)過(guò)載。這也是本課服務(wù)端退避需要少量抖動(dòng)的原因。答案的第二段實(shí)際進(jìn)入第13課FastAPI的lifespan通過(guò)ASGITransport提交請(qǐng)求。先占用唯一入站槽位斷言響應(yīng)為503且Retry-After為1釋放槽位后讓供應(yīng)商本身返回503斷言接口此時(shí)是200、best為空、三項(xiàng)failures為http_503。這使“本服務(wù)拒絕準(zhǔn)入”和“供應(yīng)商查詢失敗”兩個(gè)不同位置的503得到完整驗(yàn)證。先問(wèn)操作是否適合重試再問(wèn)異常類型QuoteFlow執(zhí)行的是讀取報(bào)價(jià)的GET本次第二次嘗試不應(yīng)創(chuàng)建訂單或重復(fù)扣款。這樣的讀取操作允許我們討論有限重試。如果把同樣代碼搬到支付POST即使客戶端沒(méi)有收到響應(yīng)服務(wù)器也可能已經(jīng)完成扣款傳輸錯(cuò)誤只說(shuō)明我們不知道結(jié)果不能證明遠(yuǎn)端沒(méi)有執(zhí)行。那種業(yè)務(wù)需要冪等鍵、結(jié)果查詢或其他明確協(xié)議本課的讀取重試不能當(dāng)作通用事務(wù)恢復(fù)方案。即便是GET也應(yīng)考慮供應(yīng)商成本、限額和服務(wù)條款。兩次嘗試意味著故障持續(xù)時(shí)每個(gè)邏輯調(diào)用的流量可能翻倍三家供應(yīng)商的查詢?cè)咀疃嗳蜨TTP本課在都可重試失敗時(shí)最多六次HTTP。固定兩次讓放大倍數(shù)有上限而總預(yù)算進(jìn)一步限制慢故障時(shí)能夠真正發(fā)出的嘗試數(shù)量。不能說(shuō)“最多兩次所以肯定會(huì)試滿兩次”第一輪已用完預(yù)算第二輪就沒(méi)有資格開(kāi)始。請(qǐng)求ID在一次供應(yīng)商查詢的所有嘗試中保持一致。實(shí)驗(yàn)服務(wù)器的flaky規(guī)則是同一個(gè)X-Request-ID第一次503、第二次正常如果每次循環(huán)新建ID服務(wù)器會(huì)把每次都視為第一次結(jié)果永遠(yuǎn)失敗。這個(gè)ID幫助關(guān)聯(lián)同一邏輯查詢的嘗試不等于已經(jīng)實(shí)現(xiàn)支付級(jí)冪等也不能僅憑一個(gè)隨機(jī)字符串就宣稱全鏈路追蹤已經(jīng)完整。錯(cuò)誤分類的目的是決定下一步而不是收集名稱HTTP503屬于當(dāng)前供應(yīng)商暫時(shí)不可用本課程允許重試。404表示這個(gè)供應(yīng)商沒(méi)有對(duì)應(yīng)資源重復(fù)同一讀取通常不能憑空產(chǎn)生SKU因此立即返回http_404。其他HTTP狀態(tài)本實(shí)現(xiàn)也直接報(bào)告不假設(shè)每一種5xx或429都適合無(wú)條件重試。若將來(lái)支持Retry-After、429或某些500應(yīng)分別為供應(yīng)商合同與等待上限寫測(cè)試而非簡(jiǎn)單擴(kuò)大一個(gè)狀態(tài)碼范圍。200之后的畸形JSON、字段類型錯(cuò)誤、負(fù)金額、響應(yīng)身份或SKU錯(cuò)配都屬于invalid_data重試不會(huì)自動(dòng)修復(fù)協(xié)議。這些錯(cuò)誤必須保留可見(jiàn)證據(jù)讓供應(yīng)商或我們修正合同。庫(kù)存不足更不是重試信號(hào)它是一份已經(jīng)讀到的有效業(yè)務(wù)結(jié)果過(guò)濾即可。否則一個(gè)庫(kù)存為1的cedar會(huì)在購(gòu)買2件時(shí)無(wú)意義地消耗兩次網(wǎng)絡(luò)操作。壞gzip是發(fā)生得更早的一種數(shù)據(jù)錯(cuò)誤HTTPX讀取響應(yīng)正文時(shí)就會(huì)拋DecodingError還沒(méi)有機(jī)會(huì)執(zhí)行response.json。它不是TransportError的子類我們?cè)赾lient.get外單獨(dú)捕獲并立即返回invalid_data絕不加入TRANSIENT集合。測(cè)試返回真實(shí)的錯(cuò)誤壓縮字節(jié)讓HTTPX自己解碼斷言boreal只訪問(wèn)一次、另外兩家仍保留、HTTP入口仍為200并驗(yàn)證響應(yīng)流和容量歸還。傳輸異常也沒(méi)有一鍋端地重試。TRANSIENT顯式列出ConnectError/ConnectTimeout、ReadError/ReadTimeout、WriteError/WriteTimeout、RemoteProtocolError、PoolTimeout。它們表達(dá)建立連接或讀取過(guò)程出現(xiàn)暫時(shí)障礙在這項(xiàng)只讀操作上允許第二次嘗試。LocalProtocolError等本地協(xié)議構(gòu)造問(wèn)題未列入常常需要改代碼而非增加流量。不能因?yàn)樗鼈兌祭^承某個(gè)HTTP異常父類就賦予相同恢復(fù)策略。對(duì)于測(cè)試刻意拋出的RuntimeError服務(wù)不會(huì)返回一個(gè)看似正常的transport_error。它由TaskGroup按程序異常傳播入站finally仍清理資源。故障分類與資源清理是兩件事我們可以拒絕掩蓋代碼錯(cuò)誤同時(shí)仍保證它不會(huì)泄漏容量。一次嘗試用完槽位退避必須站到隊(duì)伍外面for attempt in range(2)表明最多兩次。每一輪HTTP訪問(wèn)用async with self.outbound包住離開(kāi)這個(gè)上下文后才判斷是否可重試、是否還有次數(shù)最后await短暫退避。將sleep放進(jìn)Semaphore內(nèi)部會(huì)讓一個(gè)暫時(shí)什么也沒(méi)做的失敗請(qǐng)求霸占槽位阻止健康供應(yīng)商執(zhí)行。下游故障時(shí)這種浪費(fèi)尤其明顯。這里退避采用20–30毫秒隨機(jī)抖動(dòng)目的是讓同時(shí)失敗的請(qǐng)求不要完全同步重發(fā)。只有一次重試因此沒(méi)有必要建立復(fù)雜的指數(shù)退避配置框架若有更多嘗試指數(shù)增長(zhǎng)與上限會(huì)更重要但嘗試數(shù)增加也會(huì)改變整體合同。抖動(dòng)不是保證公平的調(diào)度算法它只是減少整齊重發(fā)的機(jī)會(huì)。本課保留0.8秒的絕對(duì)deadline。第一次調(diào)用、出站排隊(duì)、釋放槽位后的sleep、第二次再排隊(duì)與HTTP執(zhí)行全在_collect外層同一個(gè)timeout_at里。退避沒(méi)有另開(kāi)計(jì)時(shí)器給自己延長(zhǎng)生命。若第一次已用0.79秒20毫秒sleep可能直接被總預(yù)算取消結(jié)果是timeout而不是“為了湊滿兩次再拖延一輪”。HTTP層自己的timeout仍存在用于限制網(wǎng)絡(luò)階段它可能先拋出ReadTimeout服務(wù)按瞬態(tài)故障嘗試第二輪。但剩余總預(yù)算很短時(shí)外層取消會(huì)先到最終報(bào)告timeout。兩種超時(shí)的差別在于誰(shuí)規(guī)定邊界一個(gè)是客戶端網(wǎng)絡(luò)階段設(shè)置一個(gè)是整個(gè)業(yè)務(wù)請(qǐng)求的剩余預(yù)算。讀日志時(shí)必須同時(shí)理解它們而不是把所有等待異?；斐梢粋€(gè)原因。為什么失敗的結(jié)果仍能保持穩(wěn)定兩次503之后代碼返回最后一輪的http_503第一次ConnectError、第二次正常則返回正常報(bào)價(jià)第一次503、第二次invalid_data則立即返回invalid_data不能繼續(xù)開(kāi)第三次。每家供應(yīng)商獨(dú)立執(zhí)行這個(gè)決策atlas正常無(wú)需等待boreal決定是否重試才開(kāi)始自己的請(qǐng)求。所有供應(yīng)商完成后依然按總價(jià)、supplier排序結(jié)果不會(huì)因?yàn)榈诙握?qǐng)求恰好更快完成而改變展示規(guī)則。completed只有在供應(yīng)商得到最終值時(shí)才寫入尚在重試或退避的供應(yīng)商到總截止時(shí)刻會(huì)成為timeout。已經(jīng)成功的報(bào)價(jià)保留。這個(gè)關(guān)系讓調(diào)用方能分別判斷“目前有最低可用報(bào)價(jià)”與“仍有供應(yīng)商因預(yù)算沒(méi)有完成”避免把部分成功誤讀成已經(jīng)窮盡所有市場(chǎng)報(bào)價(jià)。隨機(jī)抖動(dòng)也提醒我們不要給測(cè)試寫死精確耗時(shí)。業(yè)務(wù)測(cè)試應(yīng)斷言次數(shù)、分類、可售結(jié)果與寬容的總時(shí)間上限性能測(cè)量則重復(fù)運(yùn)行、報(bào)告分布。本課不是把20毫秒當(dāng)成任何真實(shí)供應(yīng)商都最優(yōu)的參數(shù)而是給出可觀察、可修改、受預(yù)算約束的最小策略。分別做協(xié)議測(cè)試和真實(shí)HTTP實(shí)驗(yàn)傳輸重試另設(shè)四條回歸路徑ConnectError與ReadTimeout各自測(cè)試“首次失敗、次輪恢復(fù)”和“連續(xù)失敗兩輪”。它們都斷言boreal恰好兩次且X-Request-ID不變恢復(fù)時(shí)boreal重新進(jìn)入quotes持續(xù)失敗時(shí)報(bào)告transport_error同時(shí)atlas和cedar保留。這樣不會(huì)把503重試通過(guò)誤當(dāng)成所有瞬態(tài)異常分支也已得到覆蓋。階段測(cè)試讓MockTransport按照supplier與request_id計(jì)數(shù)。flaky第一次503、第二次返回200斷言boreal恰好兩次且兩次ID相同broken始終503斷言最多兩次malformed字段為字符串?dāng)嘌砸淮渭磇nvalid_data404同樣只一次。這組測(cè)試能發(fā)現(xiàn)把嘗試數(shù)寫成三次、每次換ID、或把數(shù)據(jù)校驗(yàn)失敗也重試的回歸。它們并未經(jīng)過(guò)網(wǎng)絡(luò)因此還需要真實(shí)實(shí)驗(yàn)。供應(yīng)商啟動(dòng)命令是python tools/provider_lab.py --port 9101。在獨(dú)立checkpoints/14環(huán)境中執(zhí)行python -m pip install -e .[test]、python -m pytest -q再啟動(dòng)python -m uvicorn quoteflow.api:app --host 127.0.0.1 --port 9100。Windows用Python3.11創(chuàng)建的.venv\Scripts\python.exeLinux用.venv/bin/python。Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9100/api/quotes-ContentType application/json-Body{quantity:2,scenario:flaky}Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9100/api/quotes-ContentType application/json-Body{quantity:2,scenario:malformed}第一條通?；謴?fù)boreal最低報(bào)價(jià)第二條保留atlas并報(bào)告boreal的invalid_data。每輪統(tǒng)計(jì)前要等供應(yīng)商active歸零再調(diào)用POST /resetGET /stats的per_supplier能提供真實(shí)請(qǐng)求數(shù)。這個(gè)控制接口只適合本機(jī)教學(xué)服務(wù)器不能把可重置全局統(tǒng)計(jì)的接口原樣公開(kāi)到真實(shí)供應(yīng)商環(huán)境。評(píng)審一個(gè)重試循環(huán)可以先做四次紙上演算第一種路徑是立即200總共一次HTTP沒(méi)有sleep沒(méi)有第二次領(lǐng)取槽位。第二種路徑是503后200先釋放首次槽位退避后重新競(jìng)爭(zhēng)第二輪成功即返回。第三種路徑是503后503兩輪結(jié)束返回http_503不存在第三輪。第四種路徑是503后在退避中到期外層timeout取消sleep收集器把未完成的boreal記為timeout。四條路徑的資源釋放位置相同退出原因卻各不相同。如果讀者把sleep挪到HTTP槽位內(nèi)部這四次紙上演算會(huì)立刻顯示第二到第四條都多占了一段無(wú)效容量。如果把deadline放到循環(huán)里面就會(huì)顯示第三條可能獲得兩份完整時(shí)間預(yù)算。如果捕獲CancelledError后continue第四條甚至可能越過(guò)總截止時(shí)刻繼續(xù)嘗試。這樣評(píng)審代碼是在追蹤一條請(qǐng)求的實(shí)際狀態(tài)變化而不是憑“用了async with所以一定安全”的印象。重試策略還需要和可觀察性配套。響應(yīng)保留request_id測(cè)試統(tǒng)計(jì)每個(gè)ID的嘗試次數(shù)真實(shí)部署若要解釋一次故障應(yīng)記錄供應(yīng)商、嘗試序號(hào)、分類與剩余預(yù)算但不應(yīng)隨意記錄用戶敏感數(shù)據(jù)。課程暫未引入一整套日志平臺(tái)并不意味著觀測(cè)不重要而是先把可解釋的請(qǐng)求身份和失敗代碼留在合同里讓后續(xù)工具有可靠基礎(chǔ)。先理解失敗再?zèng)Q定重試才能讓額外的請(qǐng)求真正服務(wù)于恢復(fù)。E14恢復(fù)一次與堅(jiān)持錯(cuò)誤只差一條規(guī)則編寫一個(gè)可控的flaky供應(yīng)商按同一個(gè)X-Request-ID首輪503、次輪正常斷言boreal進(jìn)入quotes且計(jì)數(shù)為2隨后提交malformed斷言boreal為invalid_data、計(jì)數(shù)為1、其他健康報(bào)價(jià)仍存在。不要只斷言best非空因?yàn)榧词雇耆珱](méi)有實(shí)現(xiàn)重試atlas也可能讓這個(gè)斷言通過(guò)。完整答案將在第15課的lessons/15/answer_previous.py。嘗試再說(shuō)明為何沒(méi)有給所有HTTP錯(cuò)誤重試為何退避必須在信號(hào)量外為何兩次嘗試仍共用原deadline。這三句話能講清才說(shuō)明你掌握了機(jī)制的適用邊界而不僅是會(huì)把請(qǐng)求放進(jìn)for循環(huán)。核心模塊全文附后。模型與入口保持上課合同service新增嘗試循環(huán)和明確的瞬態(tài)異常集合項(xiàng)目依舊是本地課程服務(wù)尚不具備鑒權(quán)、跨進(jìn)程限流、完備日志與線上治理體系。本課完整代碼lessons/14/answer_previous.pyE13 答案一次拒絕之后容量仍能歸還、復(fù)用。importasyncioimportsysfrompathlibimportPathimporthttpx sys.path.insert(0,str(Path(__file__).resolve().parents[2]/checkpoints/13/src))fromquoteflow.modelsimportQuoteRequestfromquoteflow.serviceimportBusy,QuoteServicefromquoteflow.apiimportcreate_appasyncdefmain():entered,releaseasyncio.Event(),asyncio.Event()asyncdefsupplier(request):entered.set()awaitrelease.wait()namerequest.url.path.rsplit(/,1)[1]returnhttpx.Response(200,jsondict(suppliername,skuKB-001,unit_cents100,shipping_cents0,stock100))asyncwithhttpx.AsyncClient(transporthttpx.MockTransport(supplier),base_urlhttp://provider)asclient:serviceQuoteService(client,inbound_limit1)firstasyncio.create_task(service.quote(QuoteRequest()))awaitentered.wait()try:awaitservice.quote(QuoteRequest())exceptBusy:passelse:raiseAssertionError(超過(guò)入站容量必須拒絕)release.set()assert(awaitfirst).bestisnotNoneassertservice.active_requests0assert(awaitservice.quote(QuoteRequest())).bestisnotNoneassertservice.active_requests0# HTTP層單獨(dú)驗(yàn)證Busy映射503供應(yīng)商503則是200響應(yīng)中的失敗項(xiàng)。asyncdefunavailable(request):returnhttpx.Response(503)appcreate_app(transporthttpx.MockTransport(unavailable))asyncwithapp.router.lifespan_context(app):sharedapp.state.service shared.inbound_limit1asyncwithhttpx.AsyncClient(transporthttpx.ASGITransport(appapp),base_urlhttp://api)asbrowser:withshared.admission():rejectedawaitbrowser.post(/api/quotes,json{quantity:2})assertrejected.status_code503assertrejected.headers[Retry-After]1recoveredawaitbrowser.post(/api/quotes,json{quantity:2})assertrecovered.status_code200datarecovered.json()assertdata[best]isNoneassertlen(data[failures])3assert{failure[code]forfailureindata[failures]}{http_503}assertshared.active_requests0print(E13 passed: rejection does not consume capacity; slot reused; HTTP 503/Retry-After verified)if__name____main__:asyncio.run(main())checkpoints/14/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/14/src/quoteflow/__init__.pyStandalone course checkpoint.checkpoints/14/src/quoteflow/models.pyfromtypingimportAnnotated,LiteralfrompydanticimportBaseModel,ConfigDict,Field SupplierLiteral[atlas,boreal,cedar]ScenarioLiteral[normal,slow,broken,malformed,flaky]NonnegativeIntAnnotated[int,Field(strictTrue,ge0)]classQuoteRequest(BaseModel):入口嚴(yán)格校驗(yàn)布爾值、字符串?dāng)?shù)量和調(diào)用方自定義 URL 均被拒絕。model_configConfigDict(extraforbid)sku:strField(defaultKB-001,min_length1,max_length64)quantity:intField(default2,strictTrue,ge1,le100)scenario:ScenarionormalclassSupplierQuote(BaseModel):供應(yīng)商是另一處不可信邊界金額分與庫(kù)存必須為非負(fù)整數(shù)。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/14/src/quoteflow/service.py每進(jìn)程一個(gè)服務(wù)實(shí)例所有 HTTP handler 共用準(zhǔn)入計(jì)數(shù)與出站信號(hào)量。importasyncioimportrandomfromcontextlibimportcontextmanagerfromtimeimportperf_counterfromuuidimportuuid4importhttpxfrompydanticimportValidationErrorfrom.modelsimportFailure,QuoteRequest,QuoteResult,RankedQuote,SupplierQuote SUPPLIERS(atlas,boreal,cedar)TRANSIENT(httpx.ConnectError,httpx.ConnectTimeout,httpx.ReadError,httpx.ReadTimeout,httpx.WriteError,httpx.WriteTimeout,httpx.RemoteProtocolError,httpx.PoolTimeout,)classBusy(Exception):入站容量已滿入口把它翻譯為 HTTP 503。classQuoteService:def__init__(self,client:httpx.AsyncClient,*,budget:float0.8,inbound_limit:int8,outbound_limit:int6):ifbudget0orinbound_limit0oroutbound_limit0:raiseValueError(budget and concurrency limits must be positive)self.clientclient self.budgetbudget self.inbound_limitinbound_limit self.active_requests0self.outboundasyncio.Semaphore(outbound_limit)contextmanagerdefadmission(self):# 檢查和遞增之間沒(méi)有 await同一事件循環(huán)的任務(wù)不會(huì)在此交錯(cuò)。# 這不是跨線程保證更不是依靠 GIL 實(shí)現(xiàn)的業(yè)務(wù)原子性。ifself.active_requestsself.inbound_limit:raiseBusy(inbound capacity exhausted)self.active_requests1try:yieldfinally:self.active_requests-1asyncdef_fetch(self,supplier:str,query:QuoteRequest,request_id:str):forattemptinrange(2):retryableFalsetry:# 每次 HTTP 嘗試才占出站槽位退避睡眠必須先釋放槽位。asyncwithself.outbound:responseawaitself.client.get(f/quotes/{supplier},params{sku:query.sku,scenario:query.scenario},headers{X-Request-ID:request_id},)ifresponse.status_code!200:failureFailure(suppliersupplier,codefhttp_{response.status_code})retryableresponse.status_code503else:try:quoteSupplierQuote.model_validate(response.json())except(ValueError,ValidationError):returnFailure(suppliersupplier,codeinvalid_data)ifquote.supplier!supplierorquote.sku!query.sku:returnFailure(suppliersupplier,codeinvalid_data)returnquoteexcepthttpx.DecodingError:# gzip 等響應(yīng)體解碼在 client.get 內(nèi)發(fā)生這是壞數(shù)據(jù)不能重試。returnFailure(suppliersupplier,codeinvalid_data)exceptTRANSIENT:failureFailure(suppliersupplier,codetransport_error)retryableTrueifnotretryableorattempt1:returnfailureawaitasyncio.sleep(0.02random.uniform(0,0.01))raiseAssertionError(unreachable)asyncdef_collect(self,query:QuoteRequest,deadline:float,started:float,request_id:str)-QuoteResult:completed{}asyncdefone(supplier):completed[supplier]awaitself._fetch(supplier,query,request_id)# 預(yù)算覆蓋信號(hào)量排隊(duì)、兩次 HTTP 嘗試和退避所有供應(yīng)商共享截止時(shí)刻。# timeout_at 只把它發(fā)起的取消轉(zhuǎn)換為 TimeoutError。# 外部 CancelledError 繼續(xù)傳播TaskGroup 先收尾所有子任務(wù)。try:asyncwithasyncio.timeout_at(deadline):asyncwithasyncio.TaskGroup()asgroup:forsupplierinSUPPLIERS:group.create_task(one(supplier))exceptTimeoutError:passreturnself._result(query,completed,started,request_id)staticmethoddef_result(query,completed,started,request_id):quotes,failures[],[]forsupplierinSUPPLIERS:valuecompleted.get(supplier)ifvalueisNone:failures.append(Failure(suppliersupplier,codetimeout))elifisinstance(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)asyncdefquote(self,query:QuoteRequest)-QuoteResult:startedperf_counter()deadlineasyncio.get_running_loop().time()self.budgetwithself.admission():returnawaitself._collect(query,deadline,started,uuid4().hex)checkpoints/14/src/quoteflow/api.pyimportosfromcontextlibimportasynccontextmanagerimporthttpxfromfastapiimportFastAPI,HTTPException,Requestfrom.modelsimportQuoteRequest,QuoteResultfrom.serviceimportBusy,QuoteServicedefcreate_app(*,transportNone)-FastAPI:asynccontextmanagerasyncdeflifespan(app):# 啟動(dòng)時(shí)創(chuàng)建一次應(yīng)用關(guān)閉時(shí)等待 aclose每次請(qǐng)求只借用客戶端。asyncwithhttpx.AsyncClient(base_urlos.getenv(QUOTEFLOW_PROVIDER_URL,http://127.0.0.1:9101),timeouthttpx.Timeout(0.8),limitshttpx.Limits(max_connections6,max_keepalive_connections6),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):try:returnawaitrequest.app.state.service.quote(query)exceptBusyasexc:raiseHTTPException(503,capacity_exhausted,headers{Retry-After:1})fromexcreturnapp appcreate_app()