Python 的 with 语句仅依赖 __enter__ 和 __exit__ 两个特殊方法：__enter__ 返回上下文对象（通常为 self），__exit__ 在退出时执行清理（如关闭连接），无需继承基类或显式注册；二者存在且行为正确即可支持 with。
为什么直接用 __enter__ 和 __exit__ 就能支持 with
Python 的 with 语句只依赖两个特殊方法：__enter__ 返回进入上下文时的对象（通常是自身），__exit__ 在退出时负责清理（比如关闭连接、回滚事务）。不需要继承任何基类，也不用注册或声明协议——只要这两个方法存在且行为合理，with 就能用。
常见错误是只实现了 __enter__ 却忘了在 __exit__ 中处理异常路径下的连接释放，导致连接泄漏；或者在 __exit__ 里盲目调用 close() 而不检查连接是否已断开，引发 AttributeError 或重复关闭报错。
-
__enter__应该返回self或一个有业务意义的对象（如游标），不要在这里做耗时操作（比如重连） -
__exit__的三个参数exc_type、exc_value、traceback必须接收，即使不使用；若想抑制异常，需显式返回True（但数据库场景极少需要） - 务必在
__exit__开头加if self._conn is None: return防止重复关闭或空引用
如何安全地管理连接生命周期和异常传播
数据库连接类的真正难点不在语法支持，而在「什么时机该关、什么错误该抛、事务怎么兜底」。比如执行 SQL 报错时，你希望自动回滚但不吞掉原始异常；连接中途断开时，__exit__ 不应再尝试关闭底层 socket。
典型陷阱：把 commit() 或 rollback() 放进 __exit__，却没区分成功/失败路径——结果正常退出时没提交，出错时又双重回滚。
- 推荐策略：只在
__exit__中做「连接级清理」（close()），事务控制交给用户显式调用commit()/rollback()，或提供独立的transaction()上下文管理器 - 如果真要在
__exit__自动回滚，仅当exc_type is not None时触发，且要捕获rollback()自身可能抛出的异常（如连接已失效） - 避免在
__exit__中重新抛出新异常（如raise ConnectionClosedError），会覆盖原始业务异常，破坏调试链路
一个最小可行的 PostgreSQL 连接类示例
以下基于 psycopg2，但逻辑适用于 MySQL（pymysql）、SQLite（sqlite3）等——关键是把驱动差异隔离在初始化和关闭环节。
import psycopg2
<p>class DBConnection:
def <strong>init</strong>(self, dsn):
self.dsn = dsn
self._conn = None</p><pre class="brush:php;toolbar:false;">def __enter__(self):
if self._conn is not None:
raise RuntimeError("Connection already open")
self._conn = psycopg2.connect(self.dsn)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if self._conn is None:
return
try:
if exc_type is not None:
self._conn.rollback()
self._conn.close()
except Exception:
pass # 关闭失败不干扰主异常流
finally:
self._conn = None
def execute(self, query, params=None):
with self._conn.cursor() as cur:
cur.execute(query, params)
return cur.fetchall()
用法：with DBConnection("dbname=test") as db: db.execute("SELECT 1")。注意这里 execute 是实例方法，不是 __enter__ 返回游标——后者会让用户误以为可直接 cur.execute()，但游标生命周期短于连接，容易踩坑。
SQLite 场景下要注意的文件锁和线程限制
SQLite 默认以 check_same_thread=True 初始化，这意味着同一个连接不能跨线程使用。如果你在 __enter__ 中创建连接，又在另一个线程里用它，会直接报 ProgrammingError: SQLite objects created in a thread can only be used in that same thread。
- 要么初始化时传
check_same_thread=False（需确保自己同步访问） - 要么改用连接池（如
sqlalchemy.pool.QueuePool），让__enter__每次取新连接 - 更稳妥的做法：在
__init__不建连接，延迟到__enter__，并在文档里明确写「该类非线程安全」
另外 SQLite 的 close() 在某些版本中不会立即释放文件锁，若紧接着想用其他进程读写同一 DB 文件，可能遇到 database is locked ——这不是上下文管理器的问题，但容易被归咎于 with 写法。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
