In SQLAlchemy 2, how do I subquery in insert .. on conflict update

Question

I want to build a insert subquery like:

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 *

How do I do something like that in ORM?

I tried

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)

But I am getting

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.

Generated SQL looks similar and seem to work fine

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

Answer 1

If I understand correctly, this query attempts to insert statement with an ON CONFLICT clause that updates the name and tags columns of the services table if there is a conflict on the name column. The tags column is updated using a subquery that concatenates the excluded tags with the new tag 'new 2', removes any duplicates, and removes any tags that are in the list of tags to remove (['new one'] in this case).

I think the following should do the trick

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)