SQLAlchemy 注册了 before_execute，回调却没有执行

已经用 event.listen(Connection, "before_execute", ...) 注册监听，SQL 正常执行，回调却没有进入；给 engine 随手加一个其他事件后，监听又开始工作。这时应检查 Connection 的事件启用路径，而不是先怀疑语句没有发送到数据库。

SQLAlchemy 2.1.6 中，Connection 类级监听可遇到这个边界：建立连接时，内部事件标志取决于 engine 是否有事件。即使类级分发器中已有回调，标志仍可能为假，执行阶段便跳过事件。2.0 系列也有相同表现，排查范围不应仅限 Python 3.13 或某个数据库驱动。

把监听注册到实际使用的 engine

如果监听只为一个 engine 服务，可以直接在该实例上注册，并在创建业务连接之前完成。下面是注册位置示意，回调内接入应用自己的观察逻辑：

from sqlalchemy import create_engine, event

engine = create_engine("sqlite://")

@event.listens_for(engine, "before_execute")
def before_execute(conn, clauseelement, multiparams, params, execution_options):
    pass

sqlite:// 只是示例地址，应保留项目实际的 engine 配置。不要把整个应用的 engine 换成临时数据库；真正需要调整的是监听目标与注册时机。

已有 Connection 会在创建时确定一部分分发状态。晚注册以后直接复用旧连接，不能作为新注册方式无效的证据。应在原有连接生命周期内归还旧连接，再取得新连接核对；不要在仍有事务执行时强行关闭所有连接。

如果应用确实需要全局 Connection 类级监听，应核对包含该修补的正式版本，或为明确管理的每个 engine 安装同一回调。给任意 engine 加一个无关空事件，只是在碰内部标志，不适合作为隐藏的长期依赖。

确认调用入口属于这个事件

before_execute 针对 SQLAlchemy 的高级执行入口；通过 exec_driver_sql() 直接执行驱动 SQL，不应期待它触发同一套高级事件，需要按接口选择游标级事件。不要因为两条代码最后都发送 SQL，就把它们当成相同回调路径。

ORM 的 do_orm_execute 又是 Session 层事件，不能机械替换来监听所有 Core 连接。异步 engine 还需按 SQLAlchemy 的异步事件适配规则选择同步侧目标，不能直接把这段同步注册方式复制到任意异步对象上。

检查时记录回调进入次数和语句类型即可，不输出绑定参数。覆盖新连接、连接池再次取出的连接，以及项目实际使用的执行入口；每次应调用多少次由应用契约决定，不能一律要求所有内部 SQL 只触发一次。

升级前后保留注册代码与依赖锁文件，用同一条执行入口检查监听是否触发。回退时还要确认同一回调没有被重复注册到 engine，否则可能把一次 SQL 误记成多次执行。