Python后端工程化与测试01：接手单文件服务，先说清楚它必须做对什么

Python后端工程化与测试01：接手单文件服务，先说清楚它必须做对什么

接手一个后端项目时，最容易产生的冲动是拆目录：把路由放进routers，把数据库放进repositories，再加一个services。文件数量增加了，我们却仍然回答不了一个更实际的问题：用户提交两件商品时，运费到底收一次还是两次？这一课先把问题留在一份文件里。我们要获得一份可以执行的业务合同，之后每次移动代码，都能知道自己有没有破坏它。

QuoteLedger接收已经选好的供应商报价。它不负责搜索供应商，也不收款。一次创建生成一张草稿报价单；调用方随后可以确认，或者取消。项目使用整数分保存金额，避免把人民币0.1元这样的十进制小数直接交给二进制浮点数运算。若单价是1299分，买两件，运费500分，总额应该是1299 * 2 + 500 = 3098分。运费属于整张报价单，不能写成(1299 + 500) * 2。

从一次用户操作画出合同

我们先列可观察行为，而不是内部函数名。POST /quotes接受sku、supplier、quantity、unit_cents、shipping_cents，返回201、Location响应头，以及新报价单。GET /quotes/{id}读取它；不存在的id返回404。POST /quotes/{id}/confirm把draft变成confirmed，cancel把draft变成cancelled。终态不能再变，重复确认也返回409。把重复确认当成成功同样可以是一种产品设计，但本课程已经选择“重复操作是冲突”，不能在重构时偷偷改掉。

Python里isinstance(True, int)的结果是True，因为bool是int的子类。JSON里的true却不代表客户买了一件商品。我们因此要求输入模型使用严格整数约束，而不能满足于“转换以后可以算”。这也是接口合同和Python语言规则不完全相同的地方。Pydantic模型负责把HTTP输入变成受约束的数据；本阶段路由继续承担计算与状态判断，后续才会把规则向领域层迁移。

返回的id默认来自UUID，created_at使用UTC时间。客户端可以显示时间，但不能根据本机时区猜测服务器保存了什么。测试更不应该断言“id必须等于某串随机值”或者“时间刚好等于现在”：先断言id存在、创建后可以读回同一张单。第05课再引入可注入的时间和id，让精确断言成为可能。

建好一个能重复运行的实验环境

以下PowerShell命令都从本专题交付目录python-backend-engineering-course执行。使用Python 3.11，先创建自己的虚拟环境；如果系统的py -3.11不可用，可用指向Python 3.11的绝对路径替换。不要把另一个课程的虚拟环境复制进来，环境目录里保存了与原机器有关的路径。

py-3.11-m venv.venv$PY=(Resolve-Path.\.venv\Scripts\python.exe).Path &$PY-m pip install-r.\checkpoints\01\requirements.txt &$PY-m uvicorn app:app--app-dir.\checkpoints\01--host 127.0.0.1--port 9200

-m让指定的Python解释器运行模块，避免终端里的pip和python来自不同环境。--app-dir把该阶段目录加入导入搜索路径；app:app左边是app.py模块，右边是模块里的ASGI应用对象。地址绑定127.0.0.1，只供本机实验。运行后应出现Uvicorn running on http://127.0.0.1:9200。保持这个终端运行，在另一个PowerShell终端发送请求：

$body= @{sku='BOOK-001';supplier='alpha';quantity=2;unit_cents=1299;shipping_cents=500}|ConvertTo-Json$quote=Invoke-RestMethod-Method Post-Uri http://127.0.0.1:9200/quotes-ContentType'application/json'-Body$body$quote|Select-Objectid,total_cents,statusInvoke-RestMethod-Method Post-Uri"http://127.0.0.1:9200/quotes/$($quote.id)/confirm"

第一条输出中的total_cents应为3098、status为draft，第二条应为confirmed。id和时间每次运行不同，不应照着文章复制。需要观察201和Location时使用Invoke-WebRequest查看StatusCode与Headers。PowerShell会对409等HTTP错误抛出异常，这不等于服务器崩溃；读响应状态再判断。结束实验用Ctrl+C停止这个服务器。

单文件里哪些地方值得警惕

打开完整源码时，先沿一次请求走：输入模型校验，生成id，计算金额，保存，序列化响应。再沿一次状态变更走：查找id，检查当前状态，修改数据。现在计算规则、HTTP状态、数据保存挤在同一模块，修改它们会互相影响。这是值得改善的耦合，却不意味着今天就该增加五层抽象。

create_app()是一个应用工厂。每调用一次都创建新的存储容器，再让内部路由函数通过闭包使用该容器。闭包表示函数可以访问定义它时所在作用域的变量：路由执行发生在工厂返回之后，容器仍然由这些函数引用着。这个设计保证测试创建两份应用时不会共享报价单；如果把容器搬到模块全局，两个app就可能读到彼此的数据。

本阶段使用内存保存报价。进程结束，数据消失；启动两个进程，两边各有一份字典。同一进程能跑通请求，并不能推出它具备持久化或者多实例一致性。我们不通过增加一个JSON文件来草率补洞，因为磁盘写入又会引入并发、部分写入和恢复问题。第06课会用真正的SQLite事务处理这些问题。

还有一个容易忽略的坏味道：错误路径比成功路径更脆弱。只试“创建→确认”，无法发现“查不到id时返回500”或者“确认后仍能取消”。409表示请求与当前资源状态冲突；404表示资源根本不存在；422表示请求输入不满足合同。三者含义不同，把它们全变成200再塞进一个失败字符串，会迫使每个客户端重复猜测。

按照Python语法读完这一条创建路径

输入模型里的Field(strict=True, ge=1, le=100)包含三个独立约束：strict禁止自动把字符串或布尔值转换为整数，ge表示大于等于，le表示小于等于。min_length与max_length约束字符串长度；ConfigDict(extra='forbid')拒绝没有声明的字段。客户把quantity拼成quantitiy时，接口应指出错误，而不是悄悄忽略它继续处理。

body.model_dump()把已经校验过的模型转成普通字典。字典字面量里的**会展开其中的键值；后面显式写入id、状态或派生字段时，要留意同名键后写会覆盖前写。我们拒绝多余字段，正是为了不让客户通过一项看似不起眼的JSON属性干预服务端决定的数据。

路由返回dict(quote)是创建浅拷贝。当前字段都是标量，因此足以防止调用方拿着同一字典引用做简单原地修改；若以后字典中包含嵌套列表，浅拷贝并不会复制内部列表。这一限制说明为什么对象设计会影响边界安全，而不是要求所有项目无差别地使用昂贵的深拷贝。

Location头也值得纳入合同。201表明资源已经创建，Location指出新资源的相对地址，客户端可以继续GET该地址。返回JSON是资源内容，响应头则表达协议元信息；二者都可能被调用方依赖。若只验证页面能显示一段JSON，就可能漏掉状态码错误、头信息缺失等兼容性问题。

最后看with lock覆盖的范围。状态变更在同一个锁内完成查找、判断与替换，防止一个进程内两个请求同时穿过draft检查。锁释放由上下文管理自动处理，即使中途抛HTTPException也会释放。此时我们已具备有限的进程内保护，但没有持久化、跨进程协调或客户端版本判断；列出具体限制，后面每次增加能力才有可验证的目标。

今天不急着做的结构调整

“重构”要求在合同不变的前提下改善内部结构。增加取消接口、扩大数量上限、修改重复确认策略，都属于行为变化，应有新的需求与测试。本课程先保留一个有局限但可理解的基线，然后一次只拆一个责任。你可以写一张改动清单：金额规则需要独立测试、时间需要可控、数据需要持久化、并发修改需要检测。清单给演化排顺序，而不是一次性实施全部愿望。

读源码时尤其关注默认值和边界：shipping_cents默认0代表调用方可以省略运费；数量没有默认值意味着调用方必须明确下单数量。响应里total_cents是派生值，它必须一直和单价、数量、运费保持一致。未来拆到领域对象时，我们会让它成为属性，减少重复保存带来的失配风险。

E01：把接手记录写成可执行断言

编写一个独立Python脚本，使用TestClient调用checkpoints/01中的应用工厂。分别创建quantity=1和quantity=2、unit_cents=1299、shipping_cents=500的报价，断言总额为1799和3098。确认第一张草稿后再次确认、再取消，两次都应为409；读取不存在的id应为404。断言失败时脚本必须以非零状态退出，不能只打印“看起来正常”。不要断言随机id的具体值，也不要启动真实服务器。完整参考答案在下一课。

assert actual == expected失败会抛出AssertionError；未捕获异常会让Python进程返回非零退出码，后面的自动化工具就能识别失败。这里先用普通脚本，是为了让你体会测试最小的价值：把脑中的“应当如此”变成计算机能拒绝的条件。

本课完整阶段源码

下面的文件构成这一课的实现。第02课仍使用同一阶段，只有测试组织方式变化；不要提前把第08课带version的请求复制到这里。

checkpoints/01/requirements.txt

fastapi==0.115.12 pydantic==2.13.5 uvicorn==0.34.2 httpx==0.28.1 pytest==8.3.5

checkpoints/01/app.py

"""单文件基线：先让合同可测试，存储与业务还耦合在路由闭包中。"""fromdatetimeimportUTC,datetimefromthreadingimportLockfromuuidimportuuid4fromfastapiimportFastAPI,HTTPException,ResponsefrompydanticimportBaseModel,ConfigDict,FieldclassCreateQuoteBody(BaseModel):model_config=ConfigDict(extra="forbid",str_strip_whitespace=True)sku:str=Field(min_length=1,max_length=64)supplier:str=Field(min_length=1,max_length=32)quantity:int=Field(strict=True,ge=1,le=100)unit_cents:int=Field(strict=True,ge=0,le=100_000_000)shipping_cents:int=Field(default=0,strict=True,ge=0,le=100_000_000)defcreate_app()->FastAPI:"""每个工厂拥有独立字典；锁保护同进程读改写，不能提供跨进程持久化。"""app=FastAPI(title="QuoteLedger baseline")quotes:dict[str,dict[str,object]]={}lock=Lock()@app.get("/health")defhealth()->dict[str,str]:return{"status":"ok"}@app.post("/quotes",status_code=201)defcreate(body:CreateQuoteBody,response:Response)->dict[str,object]:quote_id=uuid4().hexquote={"id":quote_id,**body.model_dump(),"total_cents":body.unit_cents*body.quantity+body.shipping_cents,"status":"draft","created_at":datetime.now(UTC).isoformat(),}withlock:quotes[quote_id]=quote response.headers["Location"]=f"/quotes/{quote_id}"returndict(quote)@app.get("/quotes/{quote_id}")defget(quote_id:str)->dict[str,object]:withlock:ifquote_idnotinquotes:raiseHTTPException(404,"quote not found")returndict(quotes[quote_id])defchange(quote_id:str,target:str)->dict[str,object]:withlock:ifquote_idnotinquotes:raiseHTTPException(404,"quote not found")current=quotes[quote_id]ifcurrent["status"]!="draft":raiseHTTPException(409,"terminal quote cannot change")# 替换对象而非原地改字典，避免已返回引用被后续操作悄悄改变。changed={**current,"status":target}quotes[quote_id]=changedreturndict(changed)@app.post("/quotes/{quote_id}/confirm")defconfirm(quote_id:str)->dict[str,object]:returnchange(quote_id,"confirmed")@app.post("/quotes/{quote_id}/cancel")defcancel(quote_id:str)->dict[str,object]:returnchange(quote_id,"cancelled")returnapp app=create_app()