SQLAlchemy 2中如何在INSERT...ON CONFLICT UPDATE里使用子查询？

需要实现PostgreSQL中的INSERT冲突更新逻辑：插入服务数据，当name字段冲突时，更新name和tags字段——合并新旧标签数组，同时移除指定的标签元素。目标SQL如下：

INSERT INTO services
(name, tags)
VALUES
('service 1', '{"new one"}')
ON CONFLICT (name) DO UPDATE SET 
name = EXCLUDED.name,
tags = (
    SELECT coalesce(ARRAY_AGG(x), ARRAY[]::VARCHAR[])
  FROM 
    UNNEST(EXCLUDED.tags || ARRAY['new 2']) AS x
    LEFT JOIN
    UNNEST(ARRAY['new one']) AS y
    ON x = y
  WHERE y IS NULL
)
RETURNING *

使用SQLAlchemy 2.0 ORM尝试实现时，编写了如下代码：

stmt = insert(Service).values(
    name=input.name,
)

stmt = stmt.on_conflict_do_update(
    index_elements=[Service.name],
    set_={
        Service.name: stmt.excluded.name,
        Service.tags: select(
            func.array_agg(column("t")),
        ).select_from(
            func.unnest(
                Service.tags + tags_list
            ).alias("t")
        ).outerjoin(
            func.unnest(
                remove_tags
            ).alias("r"),
            column("t") == column("r")
        ).where(column("r") == None)
    }
).returning(Service)

但执行时触发错误：

asyncpg.exceptions.AmbiguousFunctionError: function unnest(unknown) is not unique
HINT:  Could not choose a best candidate function. You might need to add explicit type casts.

生成的SQL语句如下：

INSERT INTO services (name, tags)
VALUES ($1::VARCHAR, $2::VARCHAR []) ON CONFLICT (name) DO
UPDATE
SET name = excluded.name,
    tags = (
        SELECT array_agg(t) AS array_agg_1
        FROM unnest(services.tags || $3::VARCHAR []) AS t
            LEFT OUTER JOIN unnest($4) AS r ON t = r
        WHERE r IS NULL
    )
RETURNING services.name,
    services.tags,
    services.id,
    services.created_at,
    services.updated_at

错误根源是remove_tags传入unnest时没有明确的类型标注，PostgreSQL无法确定使用哪个unnest函数重载。只需给参数添加显式类型转换，并对齐目标SQL的逻辑即可解决。

修改后的代码如下：

from sqlalchemy import cast, String

stmt = insert(Service).values(
    name=input.name,
    tags=tags_list  # 传入待插入的新标签数组
)

stmt = stmt.on_conflict_do_update(
    index_elements=[Service.name],
    set_={
        Service.name: stmt.excluded.name,
        Service.tags: select(
            func.coalesce(func.array_agg(column("t")), func.array([]).cast(String))
        ).select_from(
            func.unnest(stmt.excluded.tags + tags_list).alias("t")
        ).outerjoin(
            func.unnest(cast(remove_tags, String)).alias("r"),
            column("t") == column("r")
        ).where(column("r").is_(None))
    }
).returning(Service)

关键修改点 #

• 显式类型转换：通过cast(remove_tags, String)为remove_tags数组指定VARCHAR[]类型，消除PostgreSQL对unnest函数重载的选择歧义。
• 对齐目标SQL逻辑：将合并标签的逻辑从Service.tags + tags_list改为stmt.excluded.tags + tags_list，与目标SQL中使用EXCLUDED.tags的逻辑保持一致，确保合并的是待插入的新标签和指定数组。
• 处理空数组场景：添加func.coalesce(...)，当合并后没有剩余标签时，返回空的VARCHAR[]数组而非NULL，与目标SQL的行为完全匹配。

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