postgresql 在SQLAlchemy 2中,如何在insert中进行子查询..冲突更新时

jckbn6z7  于 2023-08-04  发布在  PostgreSQL
关注(0)|答案(1)|浏览(118)

我想构建一个插入子查询,如下所示:

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 *

字符串
在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

bq3bfh9z

bq3bfh9z1#

如果我理解正确的话,这个查询试图插入带有ON CONFLICT子句的语句,如果name列发生冲突,该子句将更新服务表的nametags列。tags列使用一个子查询进行更新,该子查询将排除的标记与新标记'new 2'连接起来,删除任何重复的标记,并删除要删除的标记列表中的任何标记(在本例中为['new one '])。
我认为以下几点应该可以

from sqlalchemy import insert, select, func
from sqlalchemy.dialects.postgresql import array_agg

stmt = insert(Service).values(
    name='service 1',
    tags=['new one']
)
stmt = stmt.on_conflict_do_update(
    index_elements=[Service.name],
    set_={
        Service.name: stmt.excluded.name,
        Service.tags: select(
            func.coalesce(array_agg('x'), [])
        ).select_from(
            func.unnest(stmt.excluded.tags + ['new 2']).alias('x')
        ).outerjoin(
            func.unnest(['new one']).alias('y'),
            'x' == 'y'
        ).where('y' == None)
    }
).returning(Service)

字符串

相关问题