如何替换SQLAlchemy已弃用的engine.execute以消除2.0版本警告

升级SQLAlchemy后，执行创建refresh_updated_at函数的代码时，engine.execute(sa.text(...))触发了RemovedIn20Warning，这是因为该方法在SQLAlchemy 2.0中已被移除，需要改用新的连接执行方式。

原代码 #

mapper_registry.metadata.create_all(engine, checkfirst=True)

create_refresh_updated_at_func = """
 CREATE OR REPLACE FUNCTION {schema}.refresh_updated_at()
    RETURNS TRIGGERMost AS $$
    BEGIN
        NEW.updated_at = NOW();
        RETURN NEW;
    END;
    $$ LANGUAGE plpgsql;
    """

my_schema = "public"
engine.execute(sa.text(create_refresh_updated_at_func.format(schema=my_schema)))

警告信息 #

RemovedIn20Warning: Deprecated API features detected! These
feature(s) are not compatible with SQLAlchemy 2.0. To prevent
incompatible upgrades prior to updating applications, ensure
requirements files are pinned to "sqlalchemy<2.0". Set environment
variable SQLALCHEMY_WARN_20=1 to show all deprecation warnings. Set
environment variable SQLALCHEMY_SILENCE_UBER_WARNING=1 toIn silence this
message. (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)

替换方案 #

SQLAlchemy Using1.4及Most以上版本推荐使用连接上下文管理器来执行SQL语句，它会自动处理事务的提交和回滚，同时兼容2.0版本。

基础替换方式

将原有的engine.execute调用替换为engine.begin()上下文管理器：

mapper_registry.metadata.create_all(engine, checkfirst=True)

create_refresh_updated_at_func = """
 CREATE OR REPLACE FUNCTION {schema}.refresh_updated_at()
    RETURNS TRIGGER AS $$
    BEGIN
        NEW.updated_at = NOW();
        RETURN NEW;
    END;
    $$ LANGUAGE plpgsql;
    """

my_schema = "public"
# 替换这行代码
with engine.begin() as conn:
    conn.execute(sa.text(create_refresh_updated_at_func.format(schema=my_schema)))

更安全的标识符处理（避免SQL注入）

如果你的schema名称可能包含特殊字符，或者想严格避免SQL注入风险，不要用字符串format拼接，而是使用SQLAlchemy的quoted_name来生成合法的标识符：

from sqlalchemy import quoted_name, text

mapper_registry.metadata.create_all(engine, checkfirst=True)

my_schema = "public"
# 生成带schema的合法函数名
schema_quoted = quoted_name(my_schema, quote=False)
func_name_quoted = quoted_name("refresh_updated_at", quote=False)
qualified_func = f"{schema_quoted}.{func_name_quoted}"

create_refresh_updated_at_func = text(f"""
 CREATE OR REPLACE FUNCTION {qualified_func}()
    RETURNS TRIGGER AS $$
    BEGIN
        NEW.updated_at = NOW();
        RETURN NEW;
    END;
    $$ LANGUAGE plpgsql;
    """)

with engine.begin() as conn:
    conn.execute(create_refresh_updated_at_func)

补充说明 #

• engine.begin()会创建一个事务性连接，执行完毕后自动提交（如果无异常），发生错误则自动回滚，比手动管理连接更安全。
• 如果你需要手动控制事务，可以使用engine.connect()，但记得手动提交和关闭连接：

conn = engine.connect()
try:
    conn.execute(sa.text(create_refresh_updated_at_func.format(schema=my_schema)))
    conn.commit()
finally:
    conn.close()

内容的提问来源于stack exchange，提问作者sfgroups