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

Question

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

我有一张 table :

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

堆栈跟踪中的结果:

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

results in the stacktrace: InvalidRequestError: Instance cannot be refreshed - it's not persistent and does not contain a full primary key.

所以我不想在这里成为一个坚持者，但这不是堆栈跟踪，这是一条错误消息。

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

首先，始终使用 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 上，我们必须使用自动增量列；我们从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 月，注意 issue 3133，针对 SQLAlchemy 1.0，添加以允许使用此值)，用于 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)

这适用于任何后端。

或者，如果需要“每个实例”系统，只需运行以下语句:

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