SQLAlchemy中Event.listens_for无法更新Session内Engine密码如何解决

问题原因说明 #

1. 你在do_orm_execute事件中打印的orm_execute_state.session.bind.url.password是Engine实例初始化时生成的静态URL属性，不会动态更新，这个打印值仅代表创建Engine时的初始密码，不能反映实际连接使用的真实密码，所以你看到的旧值属于正常现象，不代表事件逻辑失效。
2. 核心问题是SQLAlchemy默认会维护连接池，已经创建完成的数据库连接会被复用，不会每次执行查询都触发do_connect事件，自然不会每次拉取最新的认证token作为密码。

解决步骤 #

1. 常规场景（推荐）：调整连接池配置适配RDS IAM认证规则 #

AWS RDS的IAM认证token默认有效期为15分钟，只要保证连接复用周期小于token有效期即可，修改Engine创建参数：

__Engine = __connect_to_mysql_db(
    pool_recycle=300, # 每5分钟强制回收所有连接，短于token有效期即可
    pool_pre_ping=True, # 每次从连接池取连接前校验有效性，失效则重建连接
    pool_size=5, # 可根据业务实际并发调整连接数
    max_overflow=10
)

你现有的do_connect事件逻辑本身是正确的，每次新建连接时都会自动调用get_authentication_token()获取最新密码填入连接参数，不需要额外修改。

2. 极端场景：要求每次查询必须使用全新密码 #

如果业务不允许任何连接复用，要求每次查询都使用最新生成的密码，可以主动销毁连接池强制重建连接：
首先在db.py中新增连接销毁方法：

def dispose_all_connections():
    __Engine.dispose()

每次查询结束后调用该方法：

def get_information(item_id: int):
    with Session() as session:
        result = session.query(
            Item
        ).filter(
            Item.id == item_id
        ).first()
    # 销毁所有旧连接，下次查询会触发新建连接逻辑拉取新密码
    dispose_all_connections()
    return result

3. 正确验证密码更新的方法 #

不要读取静态URL的password字段，直接在do_connect事件中新增打印即可验证：

@event.listens_for(__Engine, 'do_connect')
def provide_token(dialect, conn_rec, cargs, cparams):
    new_token = get_authentication_token()
    print(f"新建连接使用token后10位: {new_token[-10:]}") # 仅打印后段避免泄露完整token
    cparams['password'] = new_token

内容的提问来源于stack exchange，提问作者This-is-patriiick