在 SQLAlchemy 中更新 PostgreSQL 数组字段

Question

考虑以下可标记模型：

from sqlalchemy import cast, Text
from sqlalchemy.dialects.postgresql import ARRAY, array


class User(Base):
    __tablename__ = 'users'

    tags = Column(ARRAY(Text), nullable=False,
                  default=cast(array([], type_=Text), ARRAY(Text)))

我似乎找不到任何关于如何更新该字段的文档。当然，我可以按照Update a PostgreSQL array using SQLAlchemy 中的建议做一些事情：

user = session.query(User).get(1)
user.tags = ['abc', 'def', 'ghi']
session.add(user)
session.commit()

但该解决方案假定设置整个数组值。

如果我只想向数组附加一个值怎么办？如果我想在一个查询中批量标记一组User 对象怎么办？我该怎么做？

Answer 1

您可以使用 SQLAlchemy text 和 PostgreSQL array_append 函数：

text('array_append(tags, :tag)')

对于smallint 类型，您可以使用PostgreSQL 和SQLAlchemy type castings：

text('array_append(tags, :tag\:\:smallint)')

TestTable.tags.contains(cast((1,), TestTable.tags.type))

例子：

---

将值附加到整数 PostgreSQL 数组：

    from sqlalchemy.orm import sessionmaker, scoped_session
    from sqlalchemy import create_engine, Column, Integer, update, text
    from sqlalchemy.dialects.postgresql import ARRAY
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

    class TestTable(Base):
        __tablename__ = 'test_table'

        id = Column(Integer, primary_key=True)
        tags = Column(ARRAY(Integer), nullable=False)

    engine = create_engine('postgresql://postgres')
    Base.metadata.create_all(bind=engine)
    DBSession = scoped_session(sessionmaker())
    DBSession.configure(bind=engine)

    DBSession.bulk_insert_mappings(
        TestTable,
        ({'id': i, 'tags': [i // 4]} for i in range(1, 11))
    )

    DBSession.execute(
        update(
            TestTable
        ).where(
            TestTable.tags.contains((1,))
        ).values(tags=text(f'array_append({TestTable.tags.name}, :tag)')),
        {'tag': 100}
    )

    DBSession.commit()

将一个值附加到一个小整数 PostgreSQL 数组：

    from sqlalchemy import SmallInteger, cast

    class TestTable(Base):
        __tablename__ = 'test_table2'

        id = Column(Integer, primary_key=True)
        tags = Column(ARRAY(SmallInteger), nullable=False)

    DBSession.execute(
        update(
            TestTable
        ).where(
            TestTable.tags.contains(cast((1,), TestTable.tags.type))
        ).values(tags=text(f'array_append({TestTable.tags.name}, :tag\:\:smallint)')),
        {'tag': 100}
    )

结果：

id |  tags   
----+---------
  1 | {0}
  2 | {0}
  3 | {0}
  8 | {2}
  9 | {2}
 10 | {2}
  4 | {1,100}
  5 | {1,100}
  6 | {1,100}
  7 | {1,100}