使用 SQLite 将内联查询结果分配给 SQLAlchemy 中的列

Question

我想从 SQLAlchemy 中的查询结果生成一个主键列。是的，我不使用自动增量是有原因的。

我有一张桌子：

-- Describe THERAPY
CREATE TABLE foo (
    pk INTEGER NOT NULL PRIMARY KEY,
    bar INTEGER
)

和 SQLAlchemy ORM 模型：

class Foo(Base):
    __tablename__ = 'foo'
    pk = Column(Integer, primary_key=True)
    bar = Column(Integer)

在 SQLite 中这没问题，我可以使用子查询进行简单的插入：

insert into foo (pk) values ((select coalesce(max(foo.pk) + 1, 1) from foo));

我不能在 SQLAlchemy 中这样做：

foo = Foo()
foo.pk = sql.select([sql.func.coalesce(sql.func.max(Foo.pk) + 1, 1)]).as_scalar()
session.add(foo)
session.commit()

stacktrace 中的结果：

InvalidRequestError: Instance <Foo at 0x224f710> cannot be refreshed - it's not  persistent and does not contain a full primary key.

将查询分配给 foo.bar 可以正常工作，并获得预期结果

foo = Foo()
foo.pk = 1
foo.bar = sql.select([sql.func.coalesce(sql.func.max(Foo.pk) + 1, 1)]).as_scalar()
session.add(foo)
session.commit()
# outputs (1, 1)
print (foo.bar, foo.pk)

Answer 1

导致堆栈跟踪： InvalidRequestError：无法刷新实例 - 它不是持久的并且不包含完整的主键。

所以我不想在这里成为一个顽固的人，但这不是堆栈跟踪，这是一条错误消息。

如果我们重现您的完整脚本，我们可以看到发生了什么。

首先，始终使用 echo=True 来查看发生了什么。我们可以看到该语句实际上按计划工作：

INSERT INTO foo (pk, bar) VALUES ((SELECT coalesce(max(foo.pk) + ?, ?) AS coalesce_1 

但是，错误随后会发生，并带有以下部分跟踪：

  File "/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/unitofwork.py", line 389, in finalize_flush_changes
    self.session._register_newly_persistent(other)
  File "/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/session.py", line 1408, in _register_newly_persistent
    instance_key = mapper._identity_key_from_state(state)
  File "/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/mapper.py", line 2285, in _identity_key_from_state
    for col in self.primary_key
  File "/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/attributes.py", line 580, in get
    value = callable_(state, passive)
  File "/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/state.py", line 423, in __call__
    self.manager.deferred_scalar_loader(self, toload)
  File "/Users/classic/dev/sqlalchemy/lib/sqlalchemy/orm/loading.py", line 597, in load_scalar_attributes
    "contain a full primary key." % state_str(state))
sqlalchemy.exc.InvalidRequestError: Instance <Foo at 0x10165d510> cannot be refreshed - it's not  persistent and does not contain a full primary key.

我们可以看到 SQLAlchemy ORM 和所有的 ORM 一样，需要知道这个新插入的对象的主键——“来自状态的身份键”。 SQLAlchemy 在这里失败得很不优雅。这可以改进。

要了解这里的全貌是相当复杂的，因为这里的现实有很多角落和缝隙，所以我将尝试将它们排除在外：

1. 通常，这里无法避免失败。 SQLite 不支持 RETURNING，所以当它像这样内联完成时，我们可以获得该值的唯一方法是通过 cursor.lastrowid。 Cursor.lastrowid 通常只给我们使用数据库的排序机制生成的值。在 MySQL 上，我们必须使用 AUTOINCREMENT 列；我们从 lastrowid 得到一个零。

2. 显然，基于一个简短的测试，最近的 pysqlite 在 cursor.lastrowid 中为我们提供了实际值。我以前从未见过 DBAPI 这样做，SQLAlchemy 需要额外的增强来区分 lastrowid 仅适用于自动增量值的数据库，以及不管它来自何处的实际为您提供值的 lastrowid。目前不可用。

3. 如果我们在支持 RETURNING 的数据库上按原样运行这个脚本，那么整个事情就可以正常工作。这是 Postgresql：

   INSERT INTO foo (pk, bar) VALUES ((SELECT coalesce(max(foo.pk) + %(max_1)s, %(param_1)s) AS coalesce_1 
   FROM foo), %(bar)s) RETURNING foo.pk
   

4. 所以现在（2014 年 7 月，添加了针对 SQLAlchemy 1.0 的注释 issue 3133 以允许使用此值），用于 ORM（这需要 PK，而 Core 则不需要需要，因为没有要跟踪的对象），要在 lastrowid 后端使用此模式，需要在 INSERT 之外调用该语句。我们可以使用 ColumnDefault 实现这一点：

   class Foo(Base):
       __tablename__ = 'foo'
       pk = Column(Integer, primary_key=True)
       bar = Column(Integer)
   
    Foo.__table__.c.pk.default = ColumnDefault(sql.select([sql.func.coalesce(sql.func.max(Foo.pk) + 1, 1)]).as_scalar())
   

我们将在哪里看到：

SELECT (SELECT coalesce(max(foo.pk) + ?, ?) AS coalesce_1 
FROM foo) AS anon_1
(1, 1)
INSERT INTO foo (pk, bar) VALUES (?, ?)
(1, None)

这适用于任何后端。

或者如果需要“per instance”系统，只需运行以下语句：

foo = Foo()
foo.pk = session.scalar(sql.select([sql.func.coalesce(sql.func.max(Foo.pk) + 1, 1)]))
session.add(foo)
session.commit()