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服务B，B返回503 Service Unavailable，A的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，导致Ctrl+C无法中断脚本；或者在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(f"Payment failed for {order_id}: {e}") send_alert("payment_failure", order_id)

表面看没问题，但当运维想用Ctrl+C中断这个长时间运行的支付批处理时，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): pass

2.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(f"User {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(f"User {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中会报NameError！e只在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": f"Missing 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_id=user_id, amount=amount) if not inventory_check(product_id): raise InventoryShortage(f"Product {product_id} out of stock") charge = process_payment(order) if not charge.success: raise PaymentFailed(f"Payment 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_info=True # 关键！记录完整堆栈 ) # 返回友好的JSON错误 return JSONResponse( status_code=500, 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_code=503, content={"code": 503, "msg": "Database temporarily unavailable, please retry"} ) # 其他数据库错误返回500 return JSONResponse( status_code=500, 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_info=True参数，这是日志记录的黄金配置——没有它，日志里只有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( endpoint=func.__name__, status_code=200 ).observe(duration) return result except Exception as e: duration = time.time() - start_time # 记录错误指标 ERROR_COUNTER.labels( endpoint=func.__name__, status_code=getattr(e, "status_code", 500), error_type=type(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( endpoint=func.__name__, status_code=getattr(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 问题速查表：高频异常场景与解决方案

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_info=True）
• logger.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(f"Data 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: #