零基础入门python67：FastAPI AI配额、审计与失败降级
一、上一篇课后练习讲解
上一篇练习围绕“AI标题、摘要和标签”。参考做法是先运行上一篇的测试，再用一个成功请求和一个失败请求验证边界；本篇在同一项目上增加新能力。
上一篇课后练习完整答案
上一篇练习已经落实到完整文件，运行下面代码可以观察本篇要求的成功和失败状态；数据库写入全部放在明确事务边界内。
答案要点：enrich_article 生成标题、摘要和标签，先 Pydantic 校验再写入；保存 prompt_version 与正文 hash 便于复现。
完整答案文件：app/ai_pipeline.py
完整参考答案文件
完整文件：app/ai_pipeline.py
defenrich_article(article,ai):result=ai.generate(title=article.title,content=article.content)article.ai_title=result["title"]article.ai_summary=result["summary"]article.ai_tags=result.get("tags",[])returnarticle验收：运行项目测试（FastAPI/Flask 使用 python -m pytest -q，Django 使用 python manage.py test），再按本文 curl 或 Docker 命令检查预期状态码。
完整参考答案文件
下面是交付项目中真实存在的完整文件 fastapi-blog/app/ai.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码；复制单个函数会丢失上下文，因此这里提供整份文件。
fromtypingimportProtocolfromfastapiimportAPIRouter,Dependsfrom.dependenciesimportget_current_userfrom.modelsimportUserfrom.schemasimportWritingRequestclassAIClient(Protocol):defsuggest(self,content:str)->dict:...classLocalAIClient:defsuggest(self,content):return{'title':content[:20],'summary':content[:80],'tags':['Python']}defget_ai_client():returnLocalAIClient()router=APIRouter(prefix='/api/ai',tags=['ai'])@router.post('/writing-assistant')defwriting_assistant(body:WritingRequest,user:User=Depends(get_current_user),client:AIClient=Depends(get_ai_client)):returnclient.suggest(body.content)任务需要 job_id 幂等；worker 重试时不能重复写同一版本审计。人工编辑标题优先级高于 AI 建议。
二、本篇要解决的真实问题
按用户记录调用次数，超过配额返回 429；成功和失败都记录事件，AI 不可用时返回原文和人工编辑提示。
三、请求是怎样走完整条链路的
客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任，排错时按这个顺序定位。
四、先看一个最小代码片段
timeout_seconds=10# 外部服务必须有超时，失败时返回可解释的降级结果这里的注释不是装饰：它说明数据从哪里来、为什么不能相信客户端，以及失败时系统应保持什么状态。
五、完整文件代码
app/ai.py（当前阶段完整文件）
fromtypingimportProtocolfromfastapiimportAPIRouter,Dependsfrom.dependenciesimportget_current_userfrom.modelsimportUserfrom.schemasimportWritingRequestclassAIClient(Protocol):defsuggest(self,content:str)->dict:...classLocalAIClient:defsuggest(self,content):return{'title':content[:20],'summary':content[:80],'tags':['Python']}defget_ai_client():returnLocalAIClient()router=APIRouter(prefix='/api/ai',tags=['ai'])@router.post('/writing-assistant')defwriting_assistant(body:WritingRequest,user:User=Depends(get_current_user),client:AIClient=Depends(get_ai_client)):returnclient.suggest(body.content)阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。
app/models.py（当前阶段完整文件）
fromdatetimeimportdatetime,timezonefromsqlalchemyimportForeignKey,String,Text,UniqueConstraintfromsqlalchemy.ormimportMapped,mapped_column,relationshipfrom.databaseimportBaseclassUser(Base):__tablename__='users'id:Mapped[int]=mapped_column(primary_key=True)email:Mapped[str]=mapped_column(String(120),unique=True,index=True)password_hash:Mapped[str]=mapped_column(String(255))created_at:Mapped[datetime]=mapped_column(default=lambda:datetime.now(timezone.utc))classArticle(Base):__tablename__='articles'id:Mapped[int]=mapped_column(primary_key=True)title:Mapped[str]=mapped_column(String(120),index=True)content:Mapped[str]=mapped_column(Text)author_id:Mapped[int]=mapped_column(ForeignKey('users.id'),index=True)created_at:Mapped[datetime]=mapped_column(default=lambda:datetime.now(timezone.utc))comments:Mapped[list['Comment']]=relationship(cascade='all, delete-orphan')classComment(Base):__tablename__='comments'id:Mapped[int]=mapped_column(primary_key=True)content:Mapped[str]=mapped_column(String(500))article_id:Mapped[int]=mapped_column(ForeignKey('articles.id'))user_id:Mapped[int]=mapped_column(ForeignKey('users.id'))classLike(Base):__tablename__='likes'__table_args__=(UniqueConstraint('article_id','user_id'),)id:Mapped[int]=mapped_column(primary_key=True)article_id:Mapped[int]=mapped_column(ForeignKey('articles.id'))user_id:Mapped[int]=mapped_column(ForeignKey('users.id'))阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。
六、安装、启动与验收
..\..\.venv\Scripts\python.exe-m pip install-r requirements.txt..\..\.venv\Scripts\python.exe-m pytest-q..\..\.venv\Scripts\python.exe-m uvicorn app.main:app--reload打开/docs，按顺序完成注册、登录、创建文章、分页查询、修改删除、评论点赞。错误请求必须看到明确状态码：校验错误 422，未登录 401，无权限 403，重复点赞 409。测试应全部通过。
七、本篇验收清单
- 能指出输入校验发生在哪个 schema。
- 能解释 Session 何时提交、何时回滚。
- 能用第二个用户验证资源隔离。
- 能复现一个失败场景并说明原因。
八、课后练习
围绕“AI配额、审计与失败降级”新增一个测试用例，写出请求、预期响应和断言；下一篇开头会给出参考实现，并继续使用本项目。
五、AI 配额、审计与失败降级
AI 接口昂贵，应按用户限制每天调用次数。课程采用 Redis 预扣、超限返还；数据库保存最终审计。
asyncdefreserve_quota(redis,user_id:int)->tuple[bool,int]:key=f"ai:quota:{user_id}:2026-08-06"used=awaitredis.incr(key)ifused==1:awaitredis.expire(key,86400)ifused>20:awaitredis.decr(key)returnFalse,0returnTrue,20-used模型超时返回请稍后重试或上一次建议，不能伪造成功。审计记录用户 id、耗时、供应商状态码和是否降级，不记录正文和密钥。验收 21 次调用得到第 21 次 429；FakeAI 超时后配额返还，Redis 不可用时只让 AI 503，文章和登录仍可用。
六、当前项目中的完整文件（对照阅读）
七、配额和审计的一致性
Redis 计数是快速限制，不是财务账本。每天结算或争议处理需要数据库审计记录，审计写失败时要告警并进入重试。配额返还也要幂等，使用 request_id 防止同一次失败被返还两次。
演练 Redis 宕机、模型 429、模型返回乱码三种故障，分别记录响应状态码、用户看到的提示和恢复方式。这三份证据比一句“已做降级”更能说明实现有效。
八、补充代码文件
九、一次完整的降级演练
- 正常调用 FakeAI，记录剩余配额和审计状态 success。
- 让 FakeAI 抛超时，接口返回 503 或缓存建议，配额返还，审计为 failed。
- 模拟 Redis 不可用，AI 直接 503，文章列表仍可读。
- 恢复依赖，重新调用成功，不能重复扣除失败请求的配额。
演练报告写明开始时间、故障注入方式、用户可见结果、日志证据和恢复动作。没有这些证据，就不能声称“实现了降级”。
本篇结束：完整模块文件
本节不是代码片段，而是本篇结束时该模块的完整版本。请先备份旧文件，再整体替换；替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理，未涉及的代码先不要自行删减。
app/ai.py
fromtypingimportProtocolfromfastapiimportAPIRouter,Dependsfrom.dependenciesimportget_current_userfrom.modelsimportUserfrom.schemasimportWritingRequestclassAIClient(Protocol):defsuggest(self,content:str)->dict:...classLocalAIClient:defsuggest(self,content):return{'title':content[:20],'summary':content[:80],'tags':['Python']}defget_ai_client():returnLocalAIClient()router=APIRouter(prefix='/api/ai',tags=['ai'])@router.post('/writing-assistant')defwriting_assistant(body:WritingRequest,user:User=Depends(get_current_user),client:AIClient=Depends(get_ai_client)):returnclient.suggest(body.content)----配置串口打印)
