在 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)))

我似乎找不到任何有关如何更新该字段的文档。当然，我可以按照使用 SQLAlchemy 更新 PostgreSQL 数组中的建议执行某些操作：

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 类型转换:

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}

Answer 2

这篇博文很好地描述了这种行为。

其要点（摘自这篇博文）：

事实证明，sqlalchemy 会话通过引用跟踪更改。

这意味着，没有创建新数组 - 引用没有改变，因为我们只是添加了它。

为了专门标记要更新的记录，

sqlalchemy

提供了

flag_modified

:

• flag_modified

  ：将实例上的属性标记为“已修改”。

一个非常基本的例子：

from enum import Enum as PyEnum

from sqlalchemy import ARRAY, Column, Enum, Integer
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.attributes import flag_modified

from my_project import models


engine = ...
TransactionSession = sessionmaker(bind=engine)

class ProcessStatusEnum(str, PyEnum):
    created = "created"
    started = "started"

class Process(Base):
    __tablename__ = "processes"

    id = Column(Integer, primary_key=True)
    states = Column(ARRAY(Enum(ProcessStatusEnum)), nullable=False, index=False, server_default="{%s}" % ProcessStatusEnum.created.value)

with TransactionSession.begin() as session:
    db_process = session.query(Process).filter(Process.id == 253).first()
    db_process.states.append(ProcessStatusEnum.started.value)  # adds 'started' to '["created"]'
    flag_modified(db_process, "states")  # Mark an attribute on an instance as ‘modified’.
    session.add(db_process)