零基础入门python61：FastAPI 后台任务和审计日志-创锋一号

零基础入门python61：FastAPI 后台任务和审计日志

一、上一篇课后练习讲解

上一篇练习围绕“安全文件上传”。参考做法是先运行上一篇的测试，再用一个成功请求和一个失败请求验证边界；本篇在同一项目上增加新能力。

上一篇课后练习完整答案

上一篇练习已经落实到完整文件，运行下面代码可以观察本篇要求的成功和失败状态；数据库写入全部放在明确事务边界内。
答案要点：上传检查 MIME、大小和随机文件名，只保存 owner 关联 id；下载前检查所有权，绝不返回服务器绝对路径。

完整答案文件：app/uploads.py

完整参考答案文件

完整文件：app/uploads.py

frompathlibimportPathimportuuid ALLOWED={"image/png","image/jpeg"}defsave_upload(content:bytes,content_type:str,root:Path)->str:ifcontent_typenotinALLOWEDorlen(content)>5_000_000:raiseValueError("file_invalid")name=f"{uuid.uuid4().hex}.bin"path=root/name;path.write_bytes(content)returnname

验收：运行项目测试（FastAPI/Flask 使用 python -m pytest -q，Django 使用 python manage.py test），再按本文 curl 或 Docker 命令检查预期状态码。

完整参考答案文件

下面是交付项目中真实存在的完整文件 fastapi-blog/app/extras.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码；复制单个函数会丢失上下文，因此这里提供整份文件。

frompathlibimportPathfromuuidimportuuid4fromfastapiimportAPIRouter,BackgroundTasks,File,HTTPException,UploadFilefromfastapi.responsesimportStreamingResponse router=APIRouter(prefix='/api/tools',tags=['tools'])UPLOAD_DIR=Path('uploads')ALLOWED_TYPES={'image/png':'.png','image/jpeg':'.jpg'}defwrite_audit_log(message:str):Path('audit.log').open('a',encoding='utf-8').write(message+'\n')@router.post('/avatar')asyncdefupload_avatar(file:UploadFile=File(...)):suffix=ALLOWED_TYPES.get(file.content_typeor'')ifnotsuffix:raiseHTTPException(415,'仅支持PNG和JPEG')content=awaitfile.read(2*1024*1024+1)iflen(content)>2*1024*1024:raiseHTTPException(413,'文件不能超过2MB')UPLOAD_DIR.mkdir(exist_ok=True)filename=f'{uuid4().hex}{suffix}'(UPLOAD_DIR/filename).write_bytes(content)return{'filename':filename,'size':len(content)}@router.post('/audit',status_code=202)defcreate_audit(message:str,tasks:BackgroundTasks):tasks.add_task(write_audit_log,message)return{'queued':True}asyncdefevent_stream():fortextin('正在分析','正在生成标题','生成完成'):yieldf'event: message\ndata:{text}\n\n'yield'event: done\ndata: [DONE]\n\n'@router.get('/stream')defstream():returnStreamingResponse(event_stream(),media_type='text/event-stream')

文件名使用随机 key，不能直接使用用户上传的原名；存储路径和数据库记录要有所有权，删除文章时清理孤儿文件。

二、本篇要解决的真实问题

响应先返回，耗时的审计记录放 BackgroundTasks；日志记录事件、用户和资源 id，不记录 token 和密码。

三、请求是怎样走完整条链路的

客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任，排错时按这个顺序定位。

四、先看一个最小代码片段

@router.get("/api/health")defhealth():return{"status":"ok"}

这里的注释不是装饰：它说明数据从哪里来、为什么不能相信客户端，以及失败时系统应保持什么状态。

五、完整文件代码

`app/extras.py`（当前阶段完整文件）

frompathlibimportPathfromuuidimportuuid4fromfastapiimportAPIRouter,BackgroundTasks,File,HTTPException,UploadFilefromfastapi.responsesimportStreamingResponse router=APIRouter(prefix='/api/tools',tags=['tools'])UPLOAD_DIR=Path('uploads')ALLOWED_TYPES={'image/png':'.png','image/jpeg':'.jpg'}defwrite_audit_log(message:str):Path('audit.log').open('a',encoding='utf-8').write(message+'\n')@router.post('/avatar')asyncdefupload_avatar(file:UploadFile=File(...)):suffix=ALLOWED_TYPES.get(file.content_typeor'')ifnotsuffix:raiseHTTPException(415,'仅支持PNG和JPEG')content=awaitfile.read(2*1024*1024+1)iflen(content)>2*1024*1024:raiseHTTPException(413,'文件不能超过2MB')UPLOAD_DIR.mkdir(exist_ok=True)filename=f'{uuid4().hex}{suffix}'(UPLOAD_DIR/filename).write_bytes(content)return{'filename':filename,'size':len(content)}@router.post('/audit',status_code=202)defcreate_audit(message:str,tasks:BackgroundTasks):tasks.add_task(write_audit_log,message)return{'queued':True}asyncdefevent_stream():fortextin('正在分析','正在生成标题','生成完成'):yieldf'event: message\ndata:{text}\n\n'yield'event: done\ndata: [DONE]\n\n'@router.get('/stream')defstream():returnStreamingResponse(event_stream(),media_type='text/event-stream')

阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。

`app/main.py`（当前阶段完整文件）

fromfastapiimportFastAPI,Requestfromfastapi.responsesimportJSONResponsefrom.importai,articles,auth,extrasfrom.databaseimportBase,engine Base.metadata.create_all(engine)app=FastAPI(title='Blog Community API',version='2.0')app.include_router(auth.router);app.include_router(articles.router);app.include_router(ai.router);app.include_router(extras.router)@app.get('/api/health')defhealth():return{'status':'ok'}@app.exception_handler(ValueError)defvalue_error_handler(request:Request,exc:ValueError):returnJSONResponse(status_code=400,content={'code':'BAD_REQUEST','message':str(exc)})

阅读方法：先找路由函数，再找它的 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 何时提交、何时回滚。
能用第二个用户验证资源隔离。
能复现一个失败场景并说明原因。

八、课后练习

围绕“后台任务和审计日志”新增一个测试用例，写出请求、预期响应和断言；下一篇开头会给出参考实现，并继续使用本项目。

五、后台任务和审计日志：响应快不等于可靠

BackgroundTasks 适合写不关键的本地日志和非关键通知；进程崩溃就会丢失，扣款、验证码等工作应写 outbox 表再由 worker 消费。

defwrite_audit_log(message:str)->None:"""记录最小审计信息，生产还应包含 actor 和 request_id。"""Path("audit.log").open("a",encoding="utf-8").write(message+"\n")@router.post("/audit",status_code=202)defcreate_audit(message:str,tasks:BackgroundTasks):tasks.add_task(write_audit_log,message)return{"queued":True}

202 表示已接受，不表示任务完成。日志不要包含密码、JWT 和完整正文；容器中输出 JSON 到 stdout，交给集中日志系统。故意让任务抛异常，观察响应已经返回但日志未写入，这正是它与消息队列的差别。

六、当前项目中的完整文件（对照阅读）

七、审计日志的最小字段

一条有用的审计记录至少有时间、操作者、动作、资源类型、资源 id、结果和 request_id。原始正文通常不需要保存，避免日志成为第二个敏感数据库。写文件时要考虑多进程并发，生产使用 stdout、JSON formatter 和集中采集。

BackgroundTasks 返回 202 后，客户端若需要知道最终结果，应提供任务 id 和查询接口，而不是假设几毫秒后一定完成。

八、补充代码文件

九、从 BackgroundTasks 过渡到队列

当任务需要重试、并发控制和跨进程执行时，接口先插入 outbox：

event=Outbox(event_type="ARTICLE_AUDIT",payload=json.dumps(data),status="pending")db.add(event)db.commit()

worker 轮询 pending 事件，成功后标记 done，失败记录 attempts 和 next_retry_at。数据库事务保证文章和事件一起提交，避免“文章已发布但没有审计任务”。本篇的 BackgroundTasks 是理解这个设计的最小起点，不要直接当作可靠队列。

本篇结束：完整模块文件

本节不是代码片段，而是本篇结束时该模块的完整版本。请先备份旧文件，再整体替换；替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理，未涉及的代码先不要自行删减。

app/extras.py

frompathlibimportPathfromuuidimportuuid4fromfastapiimportAPIRouter,BackgroundTasks,File,HTTPException,UploadFilefromfastapi.responsesimportStreamingResponse router=APIRouter(prefix='/api/tools',tags=['tools'])UPLOAD_DIR=Path('uploads')ALLOWED_TYPES={'image/png':'.png','image/jpeg':'.jpg'}defwrite_audit_log(message:str):Path('audit.log').open('a',encoding='utf-8').write(message+'\n')@router.post('/avatar')asyncdefupload_avatar(file:UploadFile=File(...)):suffix=ALLOWED_TYPES.get(file.content_typeor'')ifnotsuffix:raiseHTTPException(415,'仅支持PNG和JPEG')content=awaitfile.read(2*1024*1024+1)iflen(content)>2*1024*1024:raiseHTTPException(413,'文件不能超过2MB')UPLOAD_DIR.mkdir(exist_ok=True)filename=f'{uuid4().hex}{suffix}'(UPLOAD_DIR/filename).write_bytes(content)return{'filename':filename,'size':len(content)}@router.post('/audit',status_code=202)defcreate_audit(message:str,tasks:BackgroundTasks):tasks.add_task(write_audit_log,message)return{'queued':True}asyncdefevent_stream():fortextin('正在分析','正在生成标题','生成完成'):yieldf'event: message\ndata:{text}\n\n'yield'event: done\ndata: [DONE]\n\n'@router.get('/stream')defstream():returnStreamingResponse(event_stream(),media_type='text/event-stream')

企业官网建设流程全解析