插入到自动映射生成的 ORM 时出现 SQLAlchemy InvalidRequestError

Posted 2023-04-12

【中文标题】插入到自动映射生成的 ORM 时出现 SQLAlchemy InvalidRequestError

【英文标题】：SQLAlchemy InvalidRequestError when inserting to automap generated ORM

【发布时间】：2016-09-08 06:01:49

【问题描述】：

我正在尝试使用 SQLAlchemy 自动映射扩展来为现有数据库生成 ORM，并且无论何时都会收到 InvalidRequestError 异常（“无法刷新实例 - 它不是持久的并且不包含完整的主键。”）我尝试插入使用由时间戳和外键组成的复合主键的表中。

这里有一些重现问题的最小示例代码：

from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql.expression import text
from sqlalchemy.ext.automap import automap_base

db_schema_cmds = [
    '''CREATE TABLE users
    (
        u_id INTEGER NOT NULL,
        name TEXT NOT NULL,
        CONSTRAINT Key1 PRIMARY KEY (u_id)
    );''',
    '''CREATE TABLE posts
    (
        timestamp TEXT NOT NULL,
        text TEXT NOT NULL,
        u_id INTEGER NOT NULL,
        CONSTRAINT Key2 PRIMARY KEY (timestamp,u_id),
        CONSTRAINT users_have_posts FOREIGN KEY (u_id) REFERENCES users (u_id) ON DELETE CASCADE
    );''']

# Create a new in-memory SQLite DB and execute the schema SQL commands.
db_engine = create_engine('sqlite://')
with db_engine.connect() as db_conn:
    for cmd in db_schema_cmds:
        db_conn.execute(text(cmd))

# Use automap to reflect the DB schema and generate ORM classes.
Base = automap_base()
Base.prepare(db_engine, reflect=True)

# Create aliases for the table classes generated.
User = Base.classes.users
Post = Base.classes.posts

session_factory = sessionmaker()
session_factory.configure(bind=db_engine)

# Add a user and a post to the DB.
session = session_factory()
new_user = User(name="John")
session.add(new_user)
session.commit()
new_post = Post(users=new_user, text='this is a test', timestamp=func.now())
session.add(new_post)
session.commit()

# Verify that the insertion worked.
new_user_id = session.execute(select([User])).fetchone()['u_id']
new_post_fk_user_id = session.execute(select([Post])).fetchone()['u_id']
assert new_user_id == new_post_fk_user_id

session.close()

运行它会给出以下回溯：

Traceback (most recent call last):
  File "reproduce_InvalidRequestError.py", line 67, in <module>
    session.commit()
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\session.py", line 801, in commit
    self.transaction.commit()
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\session.py", line 392, in commit
    self._prepare_impl()
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\session.py", line 372, in _prepare_impl
    self.session.flush()
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\session.py", line 2019, in flush
    self._flush(objects)
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\session.py", line 2137, in _flush
    transaction.rollback(_capture_exception=True)
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\util\langhelpers.py", line 60, in __exit__
    compat.reraise(exc_type, exc_value, exc_tb)
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\util\compat.py", line 186, in reraise
    raise value
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\session.py", line 2107, in _flush
    flush_context.finalize_flush_changes()
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\unitofwork.py", line 395, in finalize_flush_changes
    self.session._register_newly_persistent(other)
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\session.py", line 1510, in _register_newly_persistent
    instance_key = mapper._identity_key_from_state(state)
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\mapper.py", line 2417, in _identity_key_from_state
    for col in self.primary_key
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\mapper.py", line 2417, in <listcomp>
    for col in self.primary_key
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\attributes.py", line 578, in get
    value = state._load_expired(state, passive)
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\state.py", line 474, in _load_expired
    self.manager.deferred_scalar_loader(self, toload)
  File "C:\Python\Python35\lib\site-packages\sqlalchemy\orm\loading.py", line 647, in load_scalar_attributes
    "contain a full primary key." % state_str(state))
sqlalchemy.exc.InvalidRequestError: Instance <posts at 0x45d45f8> cannot be refreshed - it's not  persistent and does not contain a full primary key.

如果我在create_engine 调用中添加echo=True 参数，我看到它正在为插入生成以下SQL。当我在 DB Browser for SQLite 中运行此 SQL 时，它运行良好。

INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
INFO sqlalchemy.engine.base.Engine SELECT users.u_id AS users_u_id, users.name AS users_name FROM users WHERE users.u_id = ?
INFO sqlalchemy.engine.base.Engine (1,)
INFO sqlalchemy.engine.base.Engine INSERT INTO posts (timestamp, text, u_id) VALUES (CURRENT_TIMESTAMP, ?, ?)
INFO sqlalchemy.engine.base.Engine ('this is a test', 1)
INFO sqlalchemy.engine.base.Engine ROLLBACK

我还尝试从Post() 中删除users 参数，而是在调用session.add(new_post) 之前添加new_user.posts_collection.append(new_post) 行，但这会导致生成相同的SQL 并发生相同的错误。

如果我用新的整数 PK 列替换复合键，一切正常。 （虽然不是一个理想的解决方案，因为我使用 automap 的原因是为了反映 现有 DB，因此最好不必修改该 DB 的架构。）

我发现了一个类似的问题SQLAlchemy InvalidRequestError when using composite foreign keys，但这似乎与在表 ORM 类中使用继承有关，解决方案取决于定义 ORM 表类，而不是反映数据库来生成它们。

编辑：我最初认为这个问题与我的复合主键包含外键这一事实有关。接受的答案表明外键实际上并不是问题的促成因素。

【问题讨论】：

感谢您提供完美的mcve！

【参考方案1】：

问题其实不是复合主键和外键，而是func.now()作为timestamp传递的，它是主键的一部分。由于 SQLAlchemy 不知道该值，因为它是在数据库中插入时生成的，因此无法执行后取；它不知道要获取什么。如果有问题的数据库支持RETURNING 或类似的，你就可以做到这一点。请参阅triggered columns 上的注释，其中描述了这种确切情况。 Defaults / SQL Expressions 中还介绍了主键值的预执行 SQL。

它与整数代理主键一起工作的原因是 SQLite 确实具有fetching the last inserted row id（整数主键列）的机制，SQLAlchemy 能够使用它。

要解决这个问题，您可以使用 Python 中生成的时间戳

In [8]: new_post = Post(users=new_user, text='this is a test',
   ...:                 timestamp=datetime.utcnow())
   ...: session.add(new_post)
   ...: session.commit()
   ...: 

另一种解决方案是在反射期间覆盖timestamp 列并提供func.now() 作为默认值。这将触发func.now() 的预执行。

   ...: # Use automap to reflect the DB schema and generate ORM classes.
   ...: Base = automap_base()
   ...: 
   ...: # Override timestamp column before reflection
   ...: class Post(Base):
   ...:     __tablename__ = 'posts'
   ...:     timestamp = Column(Text, nullable=False, primary_key=True,
   ...:                        default=func.now())
   ...: 
   ...: Base.prepare(db_engine, reflect=True)
   ...: 
   ...: # Create aliases for the table classes generated.
   ...: User = Base.classes.users
   ...: # Post has already been declared
   ...: #Post = Base.classes.posts

使用默认值，您不需要（也不应该）在创建新实例时提供timestamp

In [6]: new_post = Post(users=new_user, text='this is a test')
   ...: session.add(new_post)
   ...: session.commit()
   ...: 

【讨论】：

精彩的解释，谢谢！我将编辑我的帖子以消除有关外键是原因的猜测。