一次关于拆分、解耦和异常治理的代码进化记录一、单一职责落地痛点回顾ai_chat_service.py在一个文件里同时承担了LLM 流式调用编排会话记忆管理用户问题意图识别SQL 安全校验每次改动任何一个维度都得在这700行里小心翻找而且任何修改都有波及其他功能的风险。重构方案按单一职责原则拆分为独立模块services/ai_chat/ ├── __init__.py # 统一导出对外透明 ├── service.py # 主服务stream_chat 入口 ├── memory.py # MemoryManager记忆管理 ├── planner.py # LLMQuestionPlanner意图识别 └── sql_safety.py # SQL 安全作用域关键的细节在__init__.pyfromopenapi_server.services.ai_chat.serviceimportAIChatService,get_ai_chat_servicefromopenapi_server.services.ai_chat.memoryimportMemoryManager# ...__all__[AIChatService,get_ai_chat_service,MemoryManager,...]调用方只需要改 import 路径符号名称完全不变# 优化前fromopenapi_server.services.ai_chat_serviceimportAIChatService# 优化后fromopenapi_server.services.ai_chatimportAIChatService收益维度优化前优化后单文件行数700各模块 150 行职责耦合高度混编单一职责改动风险改一处可能波及全局边界清晰单元测试几乎不可能每个模块可独立测试二、代码去重痛点回顾uploadWechatBillData和uploadAlipayBillData两个接口除了解析 Excel/CSV 的细节不同后续的 AI 分类、分批入库、SSE 进度推送完全一致维护成本翻倍。重构方案提取公共方法_process_uploaded_bills()asyncdef_process_uploaded_bills(parsed_data:list[Bill],user:UserDep,platform_name:str,batch_size:int50,)-StreamingResponse:处理上传账单的公共逻辑AI分类 分批入库 SSE推送asyncdefevent_stream():foriinrange(0,len(parsed_data),batch_size):batchparsed_data[i:ibatch_size]# AI 分类resultsawaitclassify_bill_batch(batch)forj,rinenumerate(results):batch[j].transaction_typer.transaction_type# 入库withSession(engine)asdb_session:db_session.add_all(batch)db_session.commit()yield_sse_event({processed:processed,total:total})yield_sse_event({status:completed})returnStreamingResponse(event_stream(),media_typetext/event-stream)两个上传接口瞬间精简BillAPIRouter.post(/uploadWechatBillData)asyncdefupload_wechat_bill_data(excelData:UploadFile,user:UserDep):parsed_dataparse_wechat_bill(excelData)returnawait_process_uploaded_bills(parsed_data,user,微信)BillAPIRouter.post(/uploadAlipayBillData)asyncdefupload_alipay_bill_data(csvData:UploadFile,user:UserDep):parsed_dataparse_alipay_bill(csvData)returnawait_process_uploaded_bills(parsed_data,user,支付宝)启发去重的关键不在于机械地提取函数而在于识别业务流程的本质——两个接口的差异只在于第一步的解析后续流程完全相同。这种“模板方法”模式在业务开发中非常实用。三、模型文件拆分从混乱到有序30多个 Pydantic 模型堆在apis/models.py一个文件里每次改一个接口的模型都要翻半天。按业务域拆分apis/models/ ├── auth.py # 登录/注册/刷新Token ├── bill.py # 账单CRUD ├── analysis.py # 月度报告/分类统计/趋势 ├── budget.py # 预算 └── ai_chat.py # AI聊天__init__.py统一重导出调用方无需任何修改fromopenapi_server.apis.modelsimportBillItem# 完全没变四、异常处理统一告别散落的 HTTPException重构前每个路由都直接raise HTTPException(status_codexxx, detail...)ifbillisNone:raiseHTTPException(status_code404,detail账单不存在)ifnotuser:raiseHTTPException(status_code401,detail用户不存在)问题错误消息硬编码分散在各处同类错误的 HTTP 状态码在不同路由可能不一致无法统一做日志告警、重试等策略重构后定义异常层次结构classAppException(Exception):def__init__(self,message:str,status_code:int500):self.messagemessage self.status_codestatus_codeclassResourceNotFoundError(AppException):def__init__(self,resource_type:str,resource_id:int|strNone):messagef{resource_type}不存在(f (ID:{resource_id})ifresource_idelse)super().__init__(message,status_code404)classBillNotFoundError(ResourceNotFoundError):def__init__(self,bill_id:intNone):super().__init__(账单,bill_id)classUnauthorizedError(AppException):def__init__(self,message:str权限不足):super().__init__(message,status_code401)配合中间件统一捕获处理classExceptionHandlerMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):try:returnawaitcall_next(request)exceptAppExceptionase:logger.warning(应用异常: %s, 路径: %s,e.message,request.url.path)returnJSONResponse(status_codee.status_code,content{detail:e.message})exceptSQLAlchemyErrorase:logger.error(数据库异常: %s,str(e))returnJSONResponse(status_code500,content{detail:数据库操作失败})exceptExceptionase:logger.exception(未处理异常: %s,type(e).__name__)returnJSONResponse(status_code500,content{detail:服务器内部错误})路由层语义清晰# 重构前raiseHTTPException(status_code404,detail账单不存在)# 重构后raiseBillNotFoundError(bill_id)替换统计文件替换项数改进点bill_api.py7 处404→BillNotFoundError500→DatabaseErroruser_api.py8 处401→UnauthorizedErroranalysis_api.py6 处400→ValidationErrorbudget_api.py5 处404→BudgetNotFoundError五、两个细节5.1 语法错误修复services/ai_chat/service.py隐藏的语法错误——Prompt 字符串里的 ASCII 双引号与 Python 字符串定界符冲突# 有问题的代码禁止使用同比或环比统一写与指定对比周期相比。# ↑ 这个双引号结束了字符串# 修复后——使用中文引号禁止使用同比或环比统一写“与指定对比”周期相比。这类语法错误在动态语言中很容易被忽视直到特定路径被触发才暴露。5.2 全局变量的隐患重构前模块级全局变量在 import 时就创建实例_ai_chat_serviceAIChatService()# 模块加载时立即创建defget_ai_chat_service():return_ai_chat_service问题不必要的初始化开销测试时必须用dependency_overridesmock属于“后门”注入不利于延迟初始化和多实例管理改为lru_cache实现懒加载单例fromfunctoolsimportlru_cachelru_cache(maxsize1)defget_ai_chat_service():returnAIChatService()维度优化前优化后初始化时机import 时首次调用时测试友好度仅能 dependency_overrides可用patch直接 mock代码行数4 行3 行小结这次重构的核心原则是在保持对外接口完全不变的前提下彻底重构内部架构。改进项收益单一职责拆分可维护性大幅提升代码去重减少 200 行冗余代码统一异常处理日志可观测性提升错误语义化懒加载单例可测试性提升
后端代码优化
一次关于拆分、解耦和异常治理的代码进化记录一、单一职责落地痛点回顾ai_chat_service.py在一个文件里同时承担了LLM 流式调用编排会话记忆管理用户问题意图识别SQL 安全校验每次改动任何一个维度都得在这700行里小心翻找而且任何修改都有波及其他功能的风险。重构方案按单一职责原则拆分为独立模块services/ai_chat/ ├── __init__.py # 统一导出对外透明 ├── service.py # 主服务stream_chat 入口 ├── memory.py # MemoryManager记忆管理 ├── planner.py # LLMQuestionPlanner意图识别 └── sql_safety.py # SQL 安全作用域关键的细节在__init__.pyfromopenapi_server.services.ai_chat.serviceimportAIChatService,get_ai_chat_servicefromopenapi_server.services.ai_chat.memoryimportMemoryManager# ...__all__[AIChatService,get_ai_chat_service,MemoryManager,...]调用方只需要改 import 路径符号名称完全不变# 优化前fromopenapi_server.services.ai_chat_serviceimportAIChatService# 优化后fromopenapi_server.services.ai_chatimportAIChatService收益维度优化前优化后单文件行数700各模块 150 行职责耦合高度混编单一职责改动风险改一处可能波及全局边界清晰单元测试几乎不可能每个模块可独立测试二、代码去重痛点回顾uploadWechatBillData和uploadAlipayBillData两个接口除了解析 Excel/CSV 的细节不同后续的 AI 分类、分批入库、SSE 进度推送完全一致维护成本翻倍。重构方案提取公共方法_process_uploaded_bills()asyncdef_process_uploaded_bills(parsed_data:list[Bill],user:UserDep,platform_name:str,batch_size:int50,)-StreamingResponse:处理上传账单的公共逻辑AI分类 分批入库 SSE推送asyncdefevent_stream():foriinrange(0,len(parsed_data),batch_size):batchparsed_data[i:ibatch_size]# AI 分类resultsawaitclassify_bill_batch(batch)forj,rinenumerate(results):batch[j].transaction_typer.transaction_type# 入库withSession(engine)asdb_session:db_session.add_all(batch)db_session.commit()yield_sse_event({processed:processed,total:total})yield_sse_event({status:completed})returnStreamingResponse(event_stream(),media_typetext/event-stream)两个上传接口瞬间精简BillAPIRouter.post(/uploadWechatBillData)asyncdefupload_wechat_bill_data(excelData:UploadFile,user:UserDep):parsed_dataparse_wechat_bill(excelData)returnawait_process_uploaded_bills(parsed_data,user,微信)BillAPIRouter.post(/uploadAlipayBillData)asyncdefupload_alipay_bill_data(csvData:UploadFile,user:UserDep):parsed_dataparse_alipay_bill(csvData)returnawait_process_uploaded_bills(parsed_data,user,支付宝)启发去重的关键不在于机械地提取函数而在于识别业务流程的本质——两个接口的差异只在于第一步的解析后续流程完全相同。这种“模板方法”模式在业务开发中非常实用。三、模型文件拆分从混乱到有序30多个 Pydantic 模型堆在apis/models.py一个文件里每次改一个接口的模型都要翻半天。按业务域拆分apis/models/ ├── auth.py # 登录/注册/刷新Token ├── bill.py # 账单CRUD ├── analysis.py # 月度报告/分类统计/趋势 ├── budget.py # 预算 └── ai_chat.py # AI聊天__init__.py统一重导出调用方无需任何修改fromopenapi_server.apis.modelsimportBillItem# 完全没变四、异常处理统一告别散落的 HTTPException重构前每个路由都直接raise HTTPException(status_codexxx, detail...)ifbillisNone:raiseHTTPException(status_code404,detail账单不存在)ifnotuser:raiseHTTPException(status_code401,detail用户不存在)问题错误消息硬编码分散在各处同类错误的 HTTP 状态码在不同路由可能不一致无法统一做日志告警、重试等策略重构后定义异常层次结构classAppException(Exception):def__init__(self,message:str,status_code:int500):self.messagemessage self.status_codestatus_codeclassResourceNotFoundError(AppException):def__init__(self,resource_type:str,resource_id:int|strNone):messagef{resource_type}不存在(f (ID:{resource_id})ifresource_idelse)super().__init__(message,status_code404)classBillNotFoundError(ResourceNotFoundError):def__init__(self,bill_id:intNone):super().__init__(账单,bill_id)classUnauthorizedError(AppException):def__init__(self,message:str权限不足):super().__init__(message,status_code401)配合中间件统一捕获处理classExceptionHandlerMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):try:returnawaitcall_next(request)exceptAppExceptionase:logger.warning(应用异常: %s, 路径: %s,e.message,request.url.path)returnJSONResponse(status_codee.status_code,content{detail:e.message})exceptSQLAlchemyErrorase:logger.error(数据库异常: %s,str(e))returnJSONResponse(status_code500,content{detail:数据库操作失败})exceptExceptionase:logger.exception(未处理异常: %s,type(e).__name__)returnJSONResponse(status_code500,content{detail:服务器内部错误})路由层语义清晰# 重构前raiseHTTPException(status_code404,detail账单不存在)# 重构后raiseBillNotFoundError(bill_id)替换统计文件替换项数改进点bill_api.py7 处404→BillNotFoundError500→DatabaseErroruser_api.py8 处401→UnauthorizedErroranalysis_api.py6 处400→ValidationErrorbudget_api.py5 处404→BudgetNotFoundError五、两个细节5.1 语法错误修复services/ai_chat/service.py隐藏的语法错误——Prompt 字符串里的 ASCII 双引号与 Python 字符串定界符冲突# 有问题的代码禁止使用同比或环比统一写与指定对比周期相比。# ↑ 这个双引号结束了字符串# 修复后——使用中文引号禁止使用同比或环比统一写“与指定对比”周期相比。这类语法错误在动态语言中很容易被忽视直到特定路径被触发才暴露。5.2 全局变量的隐患重构前模块级全局变量在 import 时就创建实例_ai_chat_serviceAIChatService()# 模块加载时立即创建defget_ai_chat_service():return_ai_chat_service问题不必要的初始化开销测试时必须用dependency_overridesmock属于“后门”注入不利于延迟初始化和多实例管理改为lru_cache实现懒加载单例fromfunctoolsimportlru_cachelru_cache(maxsize1)defget_ai_chat_service():returnAIChatService()维度优化前优化后初始化时机import 时首次调用时测试友好度仅能 dependency_overrides可用patch直接 mock代码行数4 行3 行小结这次重构的核心原则是在保持对外接口完全不变的前提下彻底重构内部架构。改进项收益单一职责拆分可维护性大幅提升代码去重减少 200 行冗余代码统一异常处理日志可观测性提升错误语义化懒加载单例可测试性提升
