拓冰网站定制 
Python后端中间件专题02先让工单能独立运行——TicketFlow 的基线契约与健康检查上一课我们把“工单事实”从缓存、搜索和异步副作用中拆了出来。今天先不引入 Redis一个连create/get/update、租户隔离和重试语义都没固定的 API装再多中间件也只会把模糊放大。这次不是给 API 加一个泛泛的“健康检查”而是要跑出一份可判定的基线POST→GET→PATCH 可用/live不触发依赖探测PostgreSQL down 时/health为 503Redis 或搜索 down 时它仍以 200 报告degraded。先完成 EX-01-01/02并准备 FastAPITestClient与 200、201、404、409、503 的语义。按专题导读进入project/、激活 Python 3.11 venv、设置PYTHONPATHsrc本课健康实验只注入 fake不访问真实 PostgreSQL、Redis 或搜索服务。上一课练习答案先把上一班的两张分诊卡落成契约。答案 EX-01-01 [code]下面是一份可直接运行的故障契约表。关键不是文案而是三项的truth_affected都为False前提是创建事务已经提交。INCIDENTS{notification_delayed:{truth_affected:False,user_contract:工单 ID 仍可查通知状态显示处理中,recovery:从 Outbox pending 开始检查 relay、broker 和 worker幂等补发,},search_unavailable:{truth_affected:False,user_contract:按 ID 读写正常搜索端点显式返回 503,recovery:修复 Elasticsearch 后从 PostgreSQL 重建投影并校验身份集,},redis_timeout:{truth_affected:False,user_contract:普通详情在有界限条件下回源 PostgreSQL,recovery:修复 Redis让后续 miss 自然回填不回写业务行,},}assertset(INCIDENTS){notification_delayed,search_unavailable,redis_timeout}assertall(notitem[truth_affected]foriteminINCIDENTS.values())assertall(item[user_contract]anditem[recovery]foriteminINCIDENTS.values())print(3 incident contracts preserve committed ticket truth)运行命令python incident_contracts.py预期输出3 incident contracts preserve committed ticket truth如果 PostgreSQL 事务本身失败这个结论就不成立此时 API 必须返回失败不能凭借 Redis 里的副本宣称创建成功。答案 EX-01-02 [prose]GET /live只回答“进程是否还能接受 HTTP”因此不应做外部 I/O否则一次 Redis 抖动会让编排器不断重启本来正常的 API。GET /health回答 readinessPostgreSQL 不可用时返回 503因为事实读写已无法履约Redis 或搜索失败时返回可机器识别的degraded细节但不应把 PostgreSQL 支持的工单操作整体判死。本课把这两条契约直接放进create_app(service, health...)/live的函数体只返回常量不读取health/health只在注入探针时注册因此单元实验可以精确控制三项依赖状态生产 runtime 则注入真实探针。路由与探针解耦导入模块时不会打开连接。再把最小业务路径钉死TicketFlow 的基线 API 只需要三条业务路由POST /tickets要求X-Tenant-Id、X-Actor-Id和Idempotency-Key成功返回 201GET /tickets/{ticket_id}按租户读取不存在与跨租户都返回 404避免泄露 ID 是否存在PATCH /tickets/{ticket_id}只更新允许字段成功后version递增。同一租户的相同幂等键与相同请求指纹会重放原始工单仍返回稳定 ID如果 key 相同但 body 不同则返回 409。这比“收到重复 key 就直接返回上次结果”更严格因为它防止客户端不小心把一个 key 用于两个不同意图。应用工厂为什么不能抢走生命周期create_app(service, lifespan...)不在 import 时建数据库连接也不自己生产 repository。单元测试可以注入 SQLite repository 或内存 fake生产 runtime 则在 lifespan 里建立并关闭连接。设计理由不只是“好测”命令行、worker 和 Alembic 导入 API 模块时不应因为某个中间件不可达就崩溃。这里的路由不知道 SQLAlchemy只把 schema 转成 service 调用并将已知业务结果翻译成 HTTP。IdempotencyConflict到 409查无工单到 404未知异常不在这里被粗暴吞成 200。这种边界会让后面的缓存、限流和搜索降级都有明确插入位置。让/live与/health接受不同的盘问本课检查点先请求/live并断言 fake 探针调用次数仍为 0随后让 PostgreSQL 为 up、Redis 与搜索为 down断言/health返回 200 与statusdegraded最后只把 PostgreSQL 改成 down断言返回 503。这样 selector 验证的是标题承诺的健康行为而不是拿普通 CRUD 冒充健康验收。从project/运行python -m pytest tests/unit/test_ticket_api.py::test_liveness_and_dependency_injected_health_contract -q本地 Python 3.11 实际输出. [100%] 1 passed另一个test_create_get_and_update_ticket_through_the_http_contract仍保护 POST→GET→PATCH 与版本递增。fake 探针只证明 HTTP 决策可信远端 selector 会依次停/启本 Compose 项目的 Redis 与 PostgreSQL断言/live始终 200、Redis down 时/health200/degraded、PostgreSQL down 时/health503/not_ready并有界等待恢复。本文未运行 Docker远端证据仍为 PENDING。这四种写法会把小故障放大在/live中顺手 ping PostgreSQL依赖抖动会把活着的进程误判为需要重启。任何依赖 down 都返回 503这会让可回源 Redis、可重建搜索投影反过来决定事实 API 的 readiness。在模块 import 时创建 engine/client测试收集、Alembic 或 worker 导入都会被网络状态绑架。只测 CRUD 却声称健康检查已验收健康契约必须拥有自己的 selector 和失败分支。本课练习留给基线的一次重放实验。练习 EX-02-01 [code]写一个最小可执行实验对同一租户依次发出键request-001请求 A、同键同一请求 A、同键请求 B。断言 HTTP 状态码、前两次 ticket ID再直接查库断言tickets和outbox_events都只有 1 行。练习 EX-02-02 [prose]预测下一课将缓存只加在get之后上述三次 POST 的结果是否应改变。说明为什么“两次 GET 只有一次 repository 读取”可以证明加速生效却不能证明 Redis 成了事实源。下一站不是改写创建事务。第 03 篇会回答 EX-02-01/02只包装读取路径第一次 GET 回源、第二次 GET 命中repository 读取计数将从 2 降到 1。完整核心模块FastAPI 应用工厂Dependency-injectable FastAPI application factory with no import-time I/O.from__future__importannotationsfromtypingimportAnnotatedfromfastapiimportFastAPI,Header,HTTPException,statusfromfastapi.responsesimportJSONResponsefromticketflow.api.schemasimport(TicketCreateRequest,TicketResponse,TicketUpdateRequest,)fromticketflow.domain.ticketsimportIdempotencyConflict,TicketService TenantHeaderAnnotated[str,Header(aliasX-Tenant-Id,min_length1)]ActorHeaderAnnotated[str,Header(aliasX-Actor-Id,min_length1)]IdempotencyHeaderAnnotated[str,Header(aliasIdempotency-Key,min_length1)]defcreate_app(service:TicketService,*,healthNone,lifespanNone,)-FastAPI:Build an app around an injected service; callers choose database/network lifecycle.appFastAPI(titleTicketFlow,lifespanlifespan)app.get(/live)asyncdeflive()-dict[str,str]:Report process liveness without touching an external dependency.return{status:live}ifhealthisnotNone:app.get(/health)asyncdefdependency_health()-JSONResponse:Gate readiness on PostgreSQL while exposing optional degradation.reportawaithealth.check()returnJSONResponse(status_code(status.HTTP_200_OKifreport.readyelsestatus.HTTP_503_SERVICE_UNAVAILABLE),content{status:report.status,dependencies:report.dependencies,},)app.post(/tickets,response_modelTicketResponse,status_codestatus.HTTP_201_CREATED)defcreate_ticket(request:TicketCreateRequest,tenant_id:TenantHeader,actor_id:ActorHeader,idempotency_key:IdempotencyHeader,)-TicketResponse:try:returnTicketResponse.model_validate(service.create(tenant_id,actor_id,request.title,request.body,request.priority,idempotency_key,))exceptIdempotencyConflictaserror:raiseHTTPException(status_codestatus.HTTP_409_CONFLICT,detailstr(error))fromerrorapp.get(/tickets/{ticket_id},response_modelTicketResponse)defget_ticket(ticket_id:str,tenant_id:TenantHeader)-TicketResponse:ticketservice.get(tenant_id,ticket_id)ifticketisNone:raiseHTTPException(status_codestatus.HTTP_404_NOT_FOUND,detailticket not found)returnTicketResponse.model_validate(ticket)app.patch(/tickets/{ticket_id},response_modelTicketResponse)defupdate_ticket(ticket_id:str,request:TicketUpdateRequest,tenant_id:TenantHeader)-TicketResponse:ticketservice.update(tenant_id,ticket_id,titlerequest.title,bodyrequest.body,priorityrequest.priority,)ifticketisNone:raiseHTTPException(status_codestatus.HTTP_404_NOT_FOUND,detailticket not found)returnTicketResponse.model_validate(ticket)returnapp Python后端中间件专题02：先让工单能独立运行——TicketFlow 的基线契约与健康检查
Python后端中间件专题02先让工单能独立运行——TicketFlow 的基线契约与健康检查上一课我们把“工单事实”从缓存、搜索和异步副作用中拆了出来。今天先不引入 Redis一个连create/get/update、租户隔离和重试语义都没固定的 API装再多中间件也只会把模糊放大。这次不是给 API 加一个泛泛的“健康检查”而是要跑出一份可判定的基线POST→GET→PATCH 可用/live不触发依赖探测PostgreSQL down 时/health为 503Redis 或搜索 down 时它仍以 200 报告degraded。先完成 EX-01-01/02并准备 FastAPITestClient与 200、201、404、409、503 的语义。按专题导读进入project/、激活 Python 3.11 venv、设置PYTHONPATHsrc本课健康实验只注入 fake不访问真实 PostgreSQL、Redis 或搜索服务。上一课练习答案先把上一班的两张分诊卡落成契约。答案 EX-01-01 [code]下面是一份可直接运行的故障契约表。关键不是文案而是三项的truth_affected都为False前提是创建事务已经提交。INCIDENTS{notification_delayed:{truth_affected:False,user_contract:工单 ID 仍可查通知状态显示处理中,recovery:从 Outbox pending 开始检查 relay、broker 和 worker幂等补发,},search_unavailable:{truth_affected:False,user_contract:按 ID 读写正常搜索端点显式返回 503,recovery:修复 Elasticsearch 后从 PostgreSQL 重建投影并校验身份集,},redis_timeout:{truth_affected:False,user_contract:普通详情在有界限条件下回源 PostgreSQL,recovery:修复 Redis让后续 miss 自然回填不回写业务行,},}assertset(INCIDENTS){notification_delayed,search_unavailable,redis_timeout}assertall(notitem[truth_affected]foriteminINCIDENTS.values())assertall(item[user_contract]anditem[recovery]foriteminINCIDENTS.values())print(3 incident contracts preserve committed ticket truth)运行命令python incident_contracts.py预期输出3 incident contracts preserve committed ticket truth如果 PostgreSQL 事务本身失败这个结论就不成立此时 API 必须返回失败不能凭借 Redis 里的副本宣称创建成功。答案 EX-01-02 [prose]GET /live只回答“进程是否还能接受 HTTP”因此不应做外部 I/O否则一次 Redis 抖动会让编排器不断重启本来正常的 API。GET /health回答 readinessPostgreSQL 不可用时返回 503因为事实读写已无法履约Redis 或搜索失败时返回可机器识别的degraded细节但不应把 PostgreSQL 支持的工单操作整体判死。本课把这两条契约直接放进create_app(service, health...)/live的函数体只返回常量不读取health/health只在注入探针时注册因此单元实验可以精确控制三项依赖状态生产 runtime 则注入真实探针。路由与探针解耦导入模块时不会打开连接。再把最小业务路径钉死TicketFlow 的基线 API 只需要三条业务路由POST /tickets要求X-Tenant-Id、X-Actor-Id和Idempotency-Key成功返回 201GET /tickets/{ticket_id}按租户读取不存在与跨租户都返回 404避免泄露 ID 是否存在PATCH /tickets/{ticket_id}只更新允许字段成功后version递增。同一租户的相同幂等键与相同请求指纹会重放原始工单仍返回稳定 ID如果 key 相同但 body 不同则返回 409。这比“收到重复 key 就直接返回上次结果”更严格因为它防止客户端不小心把一个 key 用于两个不同意图。应用工厂为什么不能抢走生命周期create_app(service, lifespan...)不在 import 时建数据库连接也不自己生产 repository。单元测试可以注入 SQLite repository 或内存 fake生产 runtime 则在 lifespan 里建立并关闭连接。设计理由不只是“好测”命令行、worker 和 Alembic 导入 API 模块时不应因为某个中间件不可达就崩溃。这里的路由不知道 SQLAlchemy只把 schema 转成 service 调用并将已知业务结果翻译成 HTTP。IdempotencyConflict到 409查无工单到 404未知异常不在这里被粗暴吞成 200。这种边界会让后面的缓存、限流和搜索降级都有明确插入位置。让/live与/health接受不同的盘问本课检查点先请求/live并断言 fake 探针调用次数仍为 0随后让 PostgreSQL 为 up、Redis 与搜索为 down断言/health返回 200 与statusdegraded最后只把 PostgreSQL 改成 down断言返回 503。这样 selector 验证的是标题承诺的健康行为而不是拿普通 CRUD 冒充健康验收。从project/运行python -m pytest tests/unit/test_ticket_api.py::test_liveness_and_dependency_injected_health_contract -q本地 Python 3.11 实际输出. [100%] 1 passed另一个test_create_get_and_update_ticket_through_the_http_contract仍保护 POST→GET→PATCH 与版本递增。fake 探针只证明 HTTP 决策可信远端 selector 会依次停/启本 Compose 项目的 Redis 与 PostgreSQL断言/live始终 200、Redis down 时/health200/degraded、PostgreSQL down 时/health503/not_ready并有界等待恢复。本文未运行 Docker远端证据仍为 PENDING。这四种写法会把小故障放大在/live中顺手 ping PostgreSQL依赖抖动会把活着的进程误判为需要重启。任何依赖 down 都返回 503这会让可回源 Redis、可重建搜索投影反过来决定事实 API 的 readiness。在模块 import 时创建 engine/client测试收集、Alembic 或 worker 导入都会被网络状态绑架。只测 CRUD 却声称健康检查已验收健康契约必须拥有自己的 selector 和失败分支。本课练习留给基线的一次重放实验。练习 EX-02-01 [code]写一个最小可执行实验对同一租户依次发出键request-001请求 A、同键同一请求 A、同键请求 B。断言 HTTP 状态码、前两次 ticket ID再直接查库断言tickets和outbox_events都只有 1 行。练习 EX-02-02 [prose]预测下一课将缓存只加在get之后上述三次 POST 的结果是否应改变。说明为什么“两次 GET 只有一次 repository 读取”可以证明加速生效却不能证明 Redis 成了事实源。下一站不是改写创建事务。第 03 篇会回答 EX-02-01/02只包装读取路径第一次 GET 回源、第二次 GET 命中repository 读取计数将从 2 降到 1。完整核心模块FastAPI 应用工厂Dependency-injectable FastAPI application factory with no import-time I/O.from__future__importannotationsfromtypingimportAnnotatedfromfastapiimportFastAPI,Header,HTTPException,statusfromfastapi.responsesimportJSONResponsefromticketflow.api.schemasimport(TicketCreateRequest,TicketResponse,TicketUpdateRequest,)fromticketflow.domain.ticketsimportIdempotencyConflict,TicketService TenantHeaderAnnotated[str,Header(aliasX-Tenant-Id,min_length1)]ActorHeaderAnnotated[str,Header(aliasX-Actor-Id,min_length1)]IdempotencyHeaderAnnotated[str,Header(aliasIdempotency-Key,min_length1)]defcreate_app(service:TicketService,*,healthNone,lifespanNone,)-FastAPI:Build an app around an injected service; callers choose database/network lifecycle.appFastAPI(titleTicketFlow,lifespanlifespan)app.get(/live)asyncdeflive()-dict[str,str]:Report process liveness without touching an external dependency.return{status:live}ifhealthisnotNone:app.get(/health)asyncdefdependency_health()-JSONResponse:Gate readiness on PostgreSQL while exposing optional degradation.reportawaithealth.check()returnJSONResponse(status_code(status.HTTP_200_OKifreport.readyelsestatus.HTTP_503_SERVICE_UNAVAILABLE),content{status:report.status,dependencies:report.dependencies,},)app.post(/tickets,response_modelTicketResponse,status_codestatus.HTTP_201_CREATED)defcreate_ticket(request:TicketCreateRequest,tenant_id:TenantHeader,actor_id:ActorHeader,idempotency_key:IdempotencyHeader,)-TicketResponse:try:returnTicketResponse.model_validate(service.create(tenant_id,actor_id,request.title,request.body,request.priority,idempotency_key,))exceptIdempotencyConflictaserror:raiseHTTPException(status_codestatus.HTTP_409_CONFLICT,detailstr(error))fromerrorapp.get(/tickets/{ticket_id},response_modelTicketResponse)defget_ticket(ticket_id:str,tenant_id:TenantHeader)-TicketResponse:ticketservice.get(tenant_id,ticket_id)ifticketisNone:raiseHTTPException(status_codestatus.HTTP_404_NOT_FOUND,detailticket not found)returnTicketResponse.model_validate(ticket)app.patch(/tickets/{ticket_id},response_modelTicketResponse)defupdate_ticket(ticket_id:str,request:TicketUpdateRequest,tenant_id:TenantHeader)-TicketResponse:ticketservice.update(tenant_id,ticket_id,titlerequest.title,bodyrequest.body,priorityrequest.priority,)ifticketisNone:raiseHTTPException(status_codestatus.HTTP_404_NOT_FOUND,detailticket not found)returnTicketResponse.model_validate(ticket)returnapp