如何用SQLAlchemy优雅调用接收UUID数组参数的PostgreSQL存储过程？

• Python == 3.10
• flask-sqlalchemy == 2.5.1
• sqlalchemy == 1.4.46

在PostgreSQL中创建了名为delete_foo_by_uuids的存储过程，该过程接收UUID[]类型的输入参数，用于批量删除foo表中的数据。直接在SQL客户端调用该存储过程可以正常执行：

CALL delete_foo_by_uuids(ARRAY['b0f5a699-d080-4f9a-9e68-5d7b397dc4f5', '1026b21a-7ada-4a51-836e-c46579e1116c'])
[] completed in 72ms

当前在Flask应用中的实现方式如下，这种写法不仅不够优雅，还存在SQL注入风险：

uuids: list[UUID] = [uuid.uuid4(), uuid.uuid4()]
prepared_sql = ",".join([f"\'{_}\'' for _ in uuids])

with current_app.app_context():
    db.session.execute(f"CALL delete_foo_by_uuids(ARRAY[{prepared_sql}]::uuid[]);")

希望借助SQLAlchemy的工具简化SQL语句，实现类似f"CALL delete_foo_by_uuids({SomePgArrayClass(uuids)});"的优雅写法。

可以利用SQLAlchemy针对PostgreSQL的数组类型支持，实现参数化的存储过程调用，既优雅又能避免SQL注入：

方法1：参数化文本查询（推荐） #

直接使用text语句配合参数绑定，SQLAlchemy会自动适配PostgreSQL的UUID[]类型：

from sqlalchemy import text
import uuid

uuids: list[uuid.UUID] = [uuid.uuid4(), uuid.uuid4()]

with current_app.app_context():
    # 参数绑定会自动处理UUID数组的转换
    db.session.execute(
        text("CALL delete_foo_by_uuids(:uuids);"),
        {"uuids": uuids}
    )
    # 提交事务确保删除生效
    db.session.commit()

方法2：显式指定参数类型 #

如果需要更明确地声明参数类型，可以显式指定ARRAY(UUID())：

from sqlalchemy import text, bindparam
from sqlalchemy.dialects.postgresql import ARRAY, UUID
import uuid

uuids: list[uuid.UUID] = [uuid.uuid4(), uuid.uuid4()]

with current_app.app_context():
    stmt = text("CALL delete_foo_by_uuids(:uuids);")
    # 显式绑定参数类型
    stmt = stmt.bindparams(bindparam("uuids", type_=ARRAY(UUID())))
    db.session.execute(stmt, {"uuids": uuids})
    db.session.commit()

方法3：面向对象的存储过程调用 #

使用SQLAlchemy的func封装存储过程，调用方式更贴近代码风格：

from sqlalchemy import func
from sqlalchemy.dialects.postgresql import ARRAY, UUID
import uuid

# 封装存储过程的调用签名
delete_foo_proc = func.delete_foo_by_uuids(ARRAY(UUID()))

uuids: list[uuid.UUID] = [uuid.uuid4(), uuid.uuid4()]

with current_app.app_context():
    # 直接传入UUID列表作为参数
    db.session.execute(delete_foo_proc(uuids))
    db.session.commit()

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