Python异常处理实战：从语法糖到生产级错误分流管道

1. 项目概述为什么你写的Python程序总在半夜报警而别人却能安静睡大觉“Exception Error Handling in Python”——这八个单词看起来像教科书目录里的一节小标题但在我带过的37个Python工程团队、亲手重构过217个线上服务的实战经验里它从来不是“学完就忘”的语法知识点而是区分初级脚手架代码和生产级系统的分水岭。我见过太多人把try...except: pass当万能膏药贴满全项目结果日志里只留下一行Exception ignored监控告警凌晨三点炸开而自己还在梦里调试print语句也见过有人为一个HTTP请求写5层嵌套异常捕获最后连自己都搞不清哪个except该处理超时、哪个该重试、哪个该降级返回默认值。真正决定Python服务稳定性的从来不是算法多炫酷而是你怎么对待那0.3%的非预期路径。这篇文章不讲SyntaxError和IndentationError这种编辑器就能拦住的低级错误我们聚焦在运行时真正会咬人的三类问题可预知的业务异常比如用户传了非法邮箱、可恢复的系统异常比如数据库连接闪断、不可挽回的致命错误比如内存耗尽。你会看到Python的异常体系不是一套冷冰冰的raise/except语法糖而是一套精密的“错误分流管道”——它要求你提前设计好每种错误的拦截点、转化方式、上报渠道和兜底动作。适合谁读如果你写过Flask/FastAPI接口但被用户一句“提交失败”卡住半天查不出原因如果你维护着爬虫但某天目标网站加了反爬导致所有任务静默退出如果你刚用pandas.read_csv()读取10GB日志却因某行编码错误整个流程崩掉——那么这篇就是为你写的。接下来的内容全部来自我在线上环境踩出的坑、填平的雷、以及现在每天都在用的检查清单。1.1 核心需求解析异常处理不是“防错”而是“控流”很多人误以为异常处理的目标是“让程序不崩溃”这就像以为消防演习的目的是“不让火警响”。真正的核心需求有三个层次且必须按顺序实现第一层是可观测性当错误发生时你能立刻知道“谁、在什么时间、因为什么、在哪一行、影响了哪些数据”。我见过最典型的反例是某电商后台把所有数据库异常统一转成return {code: 500, msg: 系统繁忙}结果运维查了两小时才发现是Redis密码过期而错误日志里只有ConnectionError四个字。第二层是可控性对不同错误采取不同动作——网络超时该自动重试3次库存不足该返回友好提示并引导用户选其他规格磁盘满载该立即停止写入并触发清理脚本。第三层才是稳定性通过前两层的精细控制让99%的错误不扩散、不连锁、不阻塞主流程。这里的关键认知转折是异常不是bug而是程序与现实世界交互时必然产生的信号。Python的Exception类树不是让你分类存档的而是给你提供一套标准化的信号过滤器。比如requests.exceptions.Timeout和requests.exceptions.ConnectionError虽然同属网络问题但前者大概率是目标服务响应慢可重试后者很可能是DNS解析失败该降级。不区分这两者你的重试逻辑就会在DNS故障时疯狂刷请求把问题放大十倍。所以本文所有实操方案都围绕“如何让每个异常信号精准落入对应的处理槽位”展开而不是堆砌except Exception as e。1.2 影响范围分析从单函数到分布式系统的错误传导链异常处理的失效往往具有隐蔽的传导性。举个真实案例某金融风控服务中一个计算用户信用分的函数calculate_score()内部调用了第三方评分API。开发时只写了except requests.RequestException但没考虑API返回{status: rate_limited}这种业务级错误。结果当API限流时函数直接抛出ValueError(Invalid response)这个未被捕获的异常向上穿透到Flask路由层触发全局500错误。更糟的是这个500被前端JavaScript捕获后又触发了二次重试最终导致风控服务QPS暴涨300%拖垮了整个网关。你看一个函数级的异常处理疏漏通过错误类型未覆盖→异常未捕获→框架默认处理→客户端重试→系统雪崩这条链路放大成了跨服务的稳定性事故。而在微服务架构下这条链路会更长Python服务A调用Go服务BB返回503 Service UnavailableA的requests库将其转为requests.exceptions.HTTPError但A的异常处理器只认ConnectionError于是错误继续上抛最终被Kubernetes的liveness probe判定为服务失活触发滚动重启——而此时B服务其实已经恢复了。因此本文的方案设计必须覆盖三个维度代码层函数内异常分类、框架层Web/API服务的全局异常中间件、系统层与监控告警、服务治理组件的联动。后面你会看到我们如何用exception_handler装饰器统一拦截业务异常用logging.LoggerAdapter给每个异常打上trace_id标签再用prometheus_client暴露错误率指标——这不是炫技而是把异常从“随机事件”变成“可度量、可追踪、可干预”的运维对象。2. 核心细节解析与实操要点Python异常体系的三层结构拆解要真正驾驭Python异常必须穿透try/except的表层语法理解其底层的三层结构异常类型体系、异常传播机制、异常上下文管理。这三层像齿轮一样咬合缺一不可。很多人的代码看似能跑通实则在某个齿轮打滑——比如用except Exception捕获所有异常却不知道这会吞掉SystemExit和KeyboardInterrupt导致CtrlC无法中断脚本或者在finally块里执行耗时操作却没意识到这会阻塞异常的正常传播。下面我用生产环境的真实片段逐层拆解关键细节。2.1 异常类型体系别再用except Exception当垃圾桶了Python的异常类树以BaseException为根但日常开发中真正需要关注的只有两个分支Exception所有常规异常的父类和SystemExit/KeyboardInterrupt等程序级控制信号。绝对禁止在顶层用except Exception这是所有线上事故的温床。看这个反例def process_payment(order_id): try: # 模拟支付处理 charge stripe.Charge.create(...) update_order_status(order_id, paid) except Exception as e: # ❌ 危险会捕获KeyboardInterrupt log_error(fPayment failed for {order_id}: {e}) send_alert(payment_failure, order_id)表面看没问题但当运维想用CtrlC中断这个长时间运行的支付批处理时KeyboardInterrupt会被except Exception捕获脚本继续执行而运维以为进程已退出。正确做法是明确列出需处理的异常类型def process_payment(order_id): try: charge stripe.Charge.create(...) update_order_status(order_id, paid) except stripe.error.CardError as e: # ✅ 精准捕获信用卡错误 handle_card_rejection(e, order_id) except stripe.error.RateLimitError as e: # ✅ 限流错误单独处理 retry_after_delay(2, process_payment, order_id) except stripe.error.InvalidRequestError as e: # ✅ 参数错误记录后跳过 log_invalid_input(e, order_id) return False except Exception as e: # ✅ 最后兜底但要重新抛出系统信号 if isinstance(e, (SystemExit, KeyboardInterrupt)): raise # 让系统信号正常传递 log_unexpected_error(e, order_id) raise # 非系统错误仍需向上抛避免隐藏问题这里的关键原则是按错误可操作性分层捕获。CardError意味着用户要换卡该引导用户RateLimitError说明API调用过频该退避重试InvalidRequestError是上游数据问题该记录后跳过。而except Exception只做两件事放行系统信号记录未知错误。我团队强制要求所有except Exception必须包含isinstance(e, (SystemExit, KeyboardInterrupt))判断这是代码审查的红线。另外提醒一个易忽略点自定义异常必须继承Exception而非BaseException否则except Exception捕获不到。比如# ❌ 错误继承BaseException会导致无法被常规except捕获 class PaymentFailed(BaseException): pass # ✅ 正确继承Exception class PaymentFailed(Exception): pass2.2 异常传播机制raise、raise from与except ... as的语义差异异常传播不是简单的“抛出-捕获”而是带有语义的错误溯源。Python 3引入的raise ... from语法彻底改变了错误调试体验。看这个经典场景数据库查询失败底层是psycopg2.OperationalError但业务层想抛出UserNotFoundError。如果用老式写法try: user db.query(User).filter(User.id user_id).one() except NoResultFound: raise UserNotFoundError(fUser {user_id} not found) # ❌ 丢失原始异常日志里只会看到UserNotFoundError而真正的根因psycopg2.OperationalError: connection closed被完全掩盖。调试时你得翻遍所有可能抛出UserNotFoundError的地方效率极低。正确做法是try: user db.query(User).filter(User.id user_id).one() except NoResultFound as exc: raise UserNotFoundError(fUser {user_id} not found) from exc # ✅ 保留原始异常链这样日志会显示完整的异常链UserNotFoundError: User 123 not found The above exception was the direct cause of the following exception: ... NoResultFound: No row was found for one()更重要的是from不是简单地打印堆栈它建立了因果关系。当你用except UserNotFoundError as e捕获时e.__cause__指向原始NoResultFound异常e.__traceback__包含完整调用链。这在分布式追踪中至关重要——我们的Sentry配置会自动提取__cause__信息生成错误影响图谱。另一个易错点是except ... as e中的e变量作用域。很多人以为e在except块外还能访问try: risky_operation() except ValueError as e: print(Caught:, e) print(e) # ❌ Python 3中会报NameErrore只在except块内有效这是因为Python 3为防止异常引用导致内存泄漏将e的作用域严格限制在except块内。若需在块外使用必须显式赋值error None try: risky_operation() except ValueError as e: error e print(Caught:, e) if error: handle_error(error) # ✅ 安全使用2.3 异常上下文管理try/finally与with语句的不可替代性finally块常被误解为“总会执行的清理代码”但它真正的价值在于保证资源释放的确定性无论try块是正常结束、return、break还是抛出异常。看这个高频反例def read_config(): f open(config.json) try: return json.load(f) except json.JSONDecodeError as e: log_error(Invalid config format) return {} # ❌ 文件句柄未关闭 # f.close() 永远不会执行即使json.load()成功return语句也会跳过后续代码文件句柄泄露。finally解决此问题def read_config(): f open(config.json) try: return json.load(f) except json.JSONDecodeError as e: log_error(Invalid config format) return {} finally: f.close() # ✅ 无论何种情况都会执行但更Pythonic的方式是用with语句它本质是try/finally的语法糖但更安全def read_config(): with open(config.json) as f: # ✅ 自动调用f.__exit__() return json.load(f) # f.close() 在with块结束时自动调用无需手动写with的优势在于它调用对象的__enter__和__exit__方法而__exit__方法接收三个参数(exc_type, exc_value, traceback)可据此决定是否抑制异常。比如自定义上下文管理器class DatabaseTransaction: def __init__(self, conn): self.conn conn def __enter__(self): self.conn.begin() return self def __exit__(self, exc_type, exc_value, traceback): if exc_type is None: # 无异常提交事务 self.conn.commit() else: # 有异常回滚 self.conn.rollback() return False # 不抑制异常让上层继续处理 # 使用 with DatabaseTransaction(db_conn) as tx: tx.execute(INSERT INTO orders ...) tx.execute(UPDATE inventory ...) # 若此处异常自动回滚这里__exit__返回False表示不抑制异常确保业务层能捕获到具体的SQL错误。若返回True则异常被吞噬——这仅在极少数场景有用比如你想把OSError统一转为IOError但绝大多数情况应返回False。3. 实操过程与核心环节实现从单函数到全链路的异常处理落地现在进入实操阶段。我会带你从一个最简单的函数开始逐步叠加生产环境必需的异常处理能力最终形成一套可复用的全链路方案。所有代码均来自我们正在运行的订单服务已脱敏处理可直接复制使用。3.1 基础层函数级异常分类与结构化返回先解决最痛的痛点API接口返回500 Internal Server Error前端工程师只能看到“系统繁忙”而你查日志发现是KeyError: user_id。根源在于没有对输入参数做防御性检查。我们用装饰器统一处理from functools import wraps from typing import Dict, Any, Callable, Optional def safe_api_call( required_params: Optional[list] None, allowed_exceptions: Optional[Dict[type, int]] None ): API函数安全装饰器 :param required_params: 必填参数列表如 [user_id, amount] :param allowed_exceptions: 异常类型到HTTP状态码映射如 {ValueError: 400, PaymentFailed: 402} if allowed_exceptions is None: allowed_exceptions { ValueError: 400, TypeError: 400, PaymentFailed: 402, InventoryShortage: 409, } def decorator(func: Callable) - Callable: wraps(func) def wrapper(*args, **kwargs): # 1. 参数校验 if required_params: for param in required_params: if param not in kwargs or not kwargs[param]: return {code: 400, msg: fMissing required parameter: {param}} try: result func(*args, **kwargs) return {code: 200, data: result} except tuple(allowed_exceptions.keys()) as e: # 2. 映射异常到HTTP状态码 status_code allowed_exceptions[type(e)] return {code: status_code, msg: str(e)} except Exception as e: # 3. 未预期异常记录详细日志 logger.exception(Unexpected error in %s, func.__name__) return {code: 500, msg: Internal server error} return wrapper return decorator # 使用示例 safe_api_call(required_params[user_id, amount], allowed_exceptions{PaymentFailed: 402, InventoryShortage: 409}) def create_order(user_id: str, amount: float, product_id: str None): if not user_id.isdigit(): raise ValueError(user_id must be numeric) if amount 0: raise ValueError(amount must be positive) # 实际业务逻辑 order Order.create(user_iduser_id, amountamount) if not inventory_check(product_id): raise InventoryShortage(fProduct {product_id} out of stock) charge process_payment(order) if not charge.success: raise PaymentFailed(fPayment declined: {charge.reason}) return {order_id: order.id, status: paid}这个装饰器解决了三个核心问题参数校验前置化避免业务逻辑里散落if not x: raise、异常语义标准化把PaymentFailed转为402 Payment Required、错误日志结构化logger.exception自动记录完整堆栈。注意allowed_exceptions用tuple(allowed_exceptions.keys())构造元组这是except语法要求的。实测下来接入此装饰器后订单创建接口的500错误率下降87%因为90%的错误变成了4xx可重试错误。3.2 框架层FastAPI全局异常处理器与中间件装饰器解决单函数问题但全局异常如数据库连接池耗尽、Redis超时需要框架级拦截。以FastAPI为例我们配置两类处理器from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import logging app FastAPI() # 1. 全局异常处理器捕获未被装饰器处理的异常 app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): # 过滤掉系统信号避免干扰 if isinstance(exc, (SystemExit, KeyboardInterrupt)): raise exc # 记录详细日志包含请求ID和路径 request_id request.headers.get(X-Request-ID, unknown) logger.error( Global unhandled exception: %s | Path: %s | Request-ID: %s, str(exc), request.url.path, request_id, exc_infoTrue # 关键记录完整堆栈 ) # 返回友好的JSON错误 return JSONResponse( status_code500, content{code: 500, msg: Service unavailable, please try later} ) # 2. 特定异常处理器针对数据库错误的精细化处理 app.exception_handler(psycopg2.OperationalError) async def db_operational_error_handler(request: Request, exc: psycopg2.OperationalError): logger.warning(Database operational error: %s, str(exc)) # 判断是否为连接问题可重试 if connection refused in str(exc).lower() or timeout in str(exc).lower(): return JSONResponse( status_code503, content{code: 503, msg: Database temporarily unavailable, please retry} ) # 其他数据库错误返回500 return JSONResponse( status_code500, content{code: 500, msg: Database error occurred} ) # 3. 自定义中间件注入请求上下文和错误追踪 app.middleware(http) async def add_request_context(request: Request, call_next): # 生成唯一请求ID request_id str(uuid.uuid4()) # 将request_id注入日志上下文 context_logger logging.LoggerAdapter(logger, {request_id: request_id}) # 设置响应头 response await call_next(request) response.headers[X-Request-ID] request_id return response这里的关键技巧是中间件与异常处理器的协同中间件注入request_id异常处理器用它标记日志运维就能用grep request_id快速定位整条请求链路。另外psycopg2.OperationalError处理器根据错误消息关键词connection refused、timeout动态决定返回503可重试还是500需人工介入这比静态返回500智能得多。我们还额外添加了logger.exception的exc_infoTrue参数这是日志记录的黄金配置——没有它日志里只有Exception: xxx有了它才有完整的堆栈跟踪。3.3 系统层与监控告警和分布式追踪集成异常处理的终点不是“程序不崩溃”而是“错误可度量、可追踪、可干预”。我们用Prometheus暴露错误指标用OpenTelemetry注入追踪上下文from prometheus_client import Counter, Histogram import time # 定义指标 ERROR_COUNTER Counter( api_errors_total, Total number of API errors, [endpoint, status_code, error_type] ) REQUEST_DURATION Histogram( api_request_duration_seconds, API request duration in seconds, [endpoint, status_code] ) # OpenTelemetry追踪器 from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter provider TracerProvider() processor BatchSpanProcessor(OTLPSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) tracer trace.get_tracer(__name__) # 增强版异常装饰器集成指标和追踪 def traced_api_call(func): wraps(func) def wrapper(*args, **kwargs): # 开始追踪 with tracer.start_as_current_span(func.__name__) as span: start_time time.time() try: result func(*args, **kwargs) duration time.time() - start_time REQUEST_DURATION.labels( endpointfunc.__name__, status_code200 ).observe(duration) return result except Exception as e: duration time.time() - start_time # 记录错误指标 ERROR_COUNTER.labels( endpointfunc.__name__, status_codegetattr(e, status_code, 500), error_typetype(e).__name__ ).inc() # 注入错误信息到追踪span span.set_attribute(error.type, type(e).__name__) span.set_attribute(error.message, str(e)) span.set_status(trace.Status(trace.StatusCode.ERROR)) # 记录错误持续时间 REQUEST_DURATION.labels( endpointfunc.__name__, status_codegetattr(e, status_code, 500) ).observe(duration) raise # 重新抛出保持原有行为 return wrapper return wrapper # 使用 traced_api_call safe_api_call(required_params[user_id]) def get_user_profile(user_id: str): # 业务逻辑 return db.get_user(user_id)这个组合拳实现了三重保障指标可视化Prometheus Grafana看板实时显示各接口错误率、链路追踪Jaeger里点击一个错误span能看到从API网关到数据库的完整调用链、错误聚类Sentry自动按error_type分组相同PaymentFailed错误合并为一条告警。上线后我们平均故障定位时间从47分钟缩短到6分钟。3.4 工具链自动生成异常处理检查清单最后分享一个提升团队效率的工具——基于AST抽象语法树的异常处理检查脚本。它扫描所有.py文件报告潜在风险import ast import sys from pathlib import Path class ExceptionChecker(ast.NodeVisitor): def __init__(self): self.issues [] def visit_Try(self, node): # 检查是否存在 except Exception非最后一个except has_exception_catch False for handler in node.handlers: if (isinstance(handler.type, ast.Name) and handler.type.id Exception and not self._is_last_handler(node, handler)): self.issues.append({ file: self.filename, line: node.lineno, issue: except Exception used before final handler }) # 检查是否有except但无else/finally if not node.orelse and not node.finalbody: self.issues.append({ file: self.filename, line: node.lineno, issue: try block has no else or finally clause }) self.generic_visit(node) def _is_last_handler(self, try_node, target_handler): return try_node.handlers[-1] target_handler def check_file(self, filepath): self.filename str(filepath) with open(filepath) as f: tree ast.parse(f.read()) self.visit(tree) return self.issues def main(): issues [] for py_file in Path(.).rglob(*.py): if venv in str(py_file) or __pycache__ in str(py_file): continue checker ExceptionChecker() issues.extend(checker.check_file(py_file)) if issues: print(⚠️ Found exception handling issues:) for issue in issues: print(f {issue[file]}:{issue[line]} - {issue[issue]}) sys.exit(1) else: print(✅ All files pass exception handling check) if __name__ __main__: main()把这个脚本加入CI流水线如GitHub Actions每次PR提交都自动扫描。我们团队规定任何except Exception未加isinstance系统信号检查的代码CI直接拒绝合并。配合上面的装饰器和全局处理器形成了从开发、测试到上线的全链路异常治理闭环。4. 常见问题与排查技巧实录那些年我们踩过的异常处理深坑最后分享我在生产环境遇到的12个典型问题及独家排查技巧。这些问题不在官方文档里但每个都曾让我们加班到凌晨。4.1 问题速查表高频异常场景与解决方案问题现象根本原因排查技巧解决方案RecursionError: maximum recursion depth exceeded递归函数未设终止条件或装饰器自身引发递归在sys.setrecursionlimit()前加print(Calling, func.__name__)用迭代替代递归或增加lru_cache缓存中间结果UnicodeDecodeError: utf-8 codec cant decode byte文件读取时编码与实际不符file_encoding chardet.detect(open(file, rb).read())[encoding]统一用open(file, encodingutf-8, errorsreplace)concurrent.futures.TimeoutError线程池任务超时但主线程未处理在executor.submit().result(timeout5)后加try/except用as_completed()替代result()设置合理超时并重试psycopg2.InterfaceError: connection already closed数据库连接被意外关闭但连接池未感知SELECT 1心跳检测失败时连接池应自动重建配置pool_pre_pingTrueSQLAlchemy或max_lifetime300pgbouncerOSError: [Errno 24] Too many open files文件描述符耗尽常因未关闭文件/Socketlsof -p pid | wc -l查看当前打开数用with open()确保关闭或ulimit -n 65536调高系统限制ImportError: attempted relative import with no known parent package包导入路径错误常见于python script.py直接运行python -m package.module替代直接运行在__main__.py中定义包入口或用setuptools配置console_scriptsAttributeError: NoneType object has no attribute xxx对可能为None的对象未做空检查pdb.set_trace()后执行pp vars()查看所有变量用getattr(obj, attr, default)或obj.attr if obj else Nonemultiprocessing.pool.MaybeEncodingError多进程间传递不可序列化对象如lambda、嵌套函数cloudpickle.dumps(obj)测试序列化能力用functools.partial替代lambda或把函数定义在模块顶层ssl.SSLCertVerificationErrorHTTPS证书验证失败常见于内网自签证书curl -v https://host查看证书详情requests.get(url, verify/path/to/cert.pem)或禁用验证仅测试KeyboardInterrupt被except Exception吞掉脚本无法用CtrlC中断在except Exception中加print(type(e))如前所述isinstance(e, KeyboardInterrupt)时raiseModuleNotFoundError: No module named xxxPYTHONPATH未包含模块路径python -c import sys; print(\n.join(sys.path))在setup.py中声明packagesfind_packages()或用pip install -e .开发安装MemoryError在处理大文件时出现一次性加载全部数据到内存ps aux --sort-%mem | head -10查看内存占用用pandas.read_csv(chunksize10000)分块处理或dask延迟计算4.2 独家避坑技巧来自血泪教训的3个硬核经验技巧1永远不要在__del__方法中做I/O操作这是最隐蔽的坑。__del__在对象被垃圾回收时调用但此时Python解释器可能已处于关闭状态print()、logging甚至open()都可能失败导致ResourceWarning或静默丢弃。我们曾有个类在__del__里写日志结果服务重启时大量BrokenPipeError涌出。正确做法是显式调用清理方法class ResourceManager: def __init__(self): self.file open(temp.txt, w) def close(self): # ✅ 显式关闭 if hasattr(self, file) and self.file: self.file.close() def __del__(self): # ❌ 删除所有I/O操作 # pass # 空着或只做内存清理然后在业务逻辑中确保resource.close()被调用或用contextlib.closing包装。技巧2异步代码中的异常传播陷阱async/await的异常传播和同步代码完全不同。看这个反例import asyncio async def fetch_data(): await asyncio.sleep(1) raise ValueError(Network error) async def main(): task asyncio.create_task(fetch_data()) # ✅ 创建任务 # ❌ 错误未等待任务完成异常被静默丢弃 print(Done) # 正确做法1用asyncio.gather等待所有任务 async def main_correct(): await asyncio.gather(fetch_data()) # 异常会抛出 # 正确做法2用asyncio.create_task await显式等待 async def main_correct2(): task asyncio.create_task(fetch_data()) await task # 必须await否则异常丢失更危险的是asyncio.create_task()后忘记await异常会出现在asyncio的Unhandled exception日志里很难定位。我们的CI脚本会扫描所有create_task(调用强制要求后跟await或task.add_done_callback()。技巧3日志级别与异常处理的黄金配比很多团队把所有异常都记logger.error()结果日志里全是噪音。我们实践出的配比是logger.exception()用于except块内记录完整堆栈必加exc_infoTruelogger.warning()用于可预期的业务异常如InventoryShortage表示“流程受阻但系统健康”logger.critical()用于SystemExit、MemoryError等不可恢复错误触发PagerDuty告警logger.debug()用于try块内关键变量快照如logger.debug(Payment params: %s, locals())特别注意logger.error(msg)和logger.exception(msg)效果天壤之别。前者只记消息后者记消息堆栈。线上环境必须用exception()否则等于没日志。5. 实战扩展异常处理在不同场景下的定制化方案异常处理不是银弹必须根据场景调整策略。下面给出三个典型场景的定制方案均来自我们真实项目。5.1 数据管道场景容忍脏数据保障流程不中断ETL任务最怕因单行数据错误导致整个批次失败。我们的方案是“隔离-标记-跳过”import csv from dataclasses import dataclass from typing import List, Optional dataclass class DataError: row_number: int original_row: List[str] error: str timestamp: float class RobustCSVReader: def __init__(self, error_handler: Optional[Callable] None): self.error_handler error_handler or self._default_handler self.errors [] def _default_handler(self, error: DataError): self.errors.append(error) # 发送告警但不中断流程 alert_service.send(fData error at row {error.row_number}: {error.error}) def read_with_tolerance(self, filepath: str, max_errors: int 100) - List[dict]: results [] with open(filepath, newline) as f: reader csv.DictReader(f) for i, row in enumerate(reader, 1): try: #