SQLAlchemy + SQLite 中的插入操作因 NULL 标识键错误而失败

Question

tldr; 在尝试使用 ORM 映射器 class 将记录插入到名为 json_schema 的 table 中后，我收到了错误：sqlalchemy.orm.exc.FlushError: Instance <JsonSchema at 0x10657f730> has a NULL identity key。该错误的一部分显示为 “还要确保此 flush() 不会在不适当的时间发生，例如在 load() 事件中。”，我不确定是什么“不适当的时间”是指在这种情况下，或者它如何进一步帮助我解决问题。

在 sqlite 数据库中，我有一个名为 json_schema 的 table，看起来像这样。 table中的主键是由(lower(hex(randomblob(4)))).

生成的4字节十六进制文本字符串

CREATE TABLE json_schema (
        id TEXT DEFAULT (lower(hex(randomblob(4)))) NOT NULL, 
        body JSON NOT NULL, 
        date_added DATETIME NOT NULL, 
        title TEXT NOT NULL, 
        PRIMARY KEY (id)
);

我试图在以下代码块中插入一行

# this function inserts one json schema into the json_schemas table for each
# file present in the json_schemas_folder
def load_json_schemas(session, json_schemas_folder):

    for obj in json_schemas_folder.glob('**/*'):

        if obj.is_file() and obj.suffix == '.json':
            
            title = obj.stem.split('_')[0]
            date_added_string = obj.stem.split('_')[1]
            date_added = datetime.strptime(date_added_string, '%Y-%m-%d') 
            body = obj.read_text()

            new_row = JsonSchema( # JsonSchema is an ORM-mapper class to the json_schema table
                title = title,
                body = body,
                date_added = date_added
            )
            
            # Added row to session here
            session.add(new_row)

engine = create_engine('sqlite:///my.db', echo = True)
Session = sessionmaker(bind=engine)
session = Session()

load_json_schemas(session, Path("../data/json_schemas"))
# session.flush() #<-uncommenting this does not resolve the error.
session.commit()

问题：当我执行这个脚本时，我遇到了以下错误（在这个问题的 tldr 部分前面提到过）：

sqlalchemy.orm.exc.FlushError: 

Instance <JsonSchema at 0x10657f730> has a NULL identity key.  
If this is an auto-generated value, check that the database table allows generation 
of new primary key values, and that the mapped Column object is configured to expect 
these generated values.  Ensure also that this flush() is not occurring at an inappropriate 
time, such as within a load() event.

我检查了这个错误中提到的第一个问题 – “检查数据库 table 是否允许生成新的主键值” – 通过测试插入 INSERT（其中未指定 id）。这有效，所以不是错误的来源。

sqlite> insert into json_schema(body,date_added,title) values ('test','test','test');
sqlite> select * from json_schema;
cee94fc1|test|test|test

接下来，我检查了 ORM “映射的列对象是否配置为期望这些生成的值。” class。在下面的代码片段中，您可以看到 JsonSchema 继承的 id 列确实有 server_default 集，这让我相信这一点也已经得到解决。

@declarative_mixin
class DocumentMetadata:

    id = Column(Text, nullable=False, primary_key=True, server_default=text('(lower(hex(randomblob(4))))'))
    body = Column(JSON, nullable=False)
    date_added = Column(DATETIME, nullable=False)

    def __repr__(self):
        return f"<{self.__class__.__name__}{self.__dict__}>"

    @declared_attr
    def __tablename__(cls):
        return re.sub(r'(?<!^)(?=[A-Z])', '_', cls.__name__).lower()

class JsonSchema(Base, DocumentMetadata):
    title = Column(Text, nullable=False)

最后，错误显示为 “还要确保此 flush() 不会在不适当的时间发生，例如在 load() 事件中。” 我如何确定如果 flush() 发生在“不恰当的时间”？

Answer 1

SQLAlchemy does not yet support RETURNING for SQLite，并且这种情况下的主键不由 AUTOINCREMENT 处理。在这种情况下，SQLAlchemy 必须自己执行 default-generating 函数并显式插入生成的值。

要实现这一点，要么：

将列定义中的 server_default=text('(lower(hex(randomblob(4))))') 更改为 default=text('(lower(hex(randomblob(4))))')
- table 定义将不再包含 DEFAULT 子句
将 default=text('(lower(hex(randomblob(4))))') 添加到列定义中，将 server_default 保留在原位
- table 将保留 DEFAULT 子句，尽管 SQLAlchemy 将始终覆盖它。

这记录在 Fetching Server-Generated Defaults, in particular in the Case 4: primary key, RETURNING or equivalent is not supported 部分。

第二种方法的新ORM-mapperclass：

@declarative_mixin
class DocumentMetadata:

    # this is the id column for the second approach. 
    #notice that `default` and `server_default` are both set
    id = Column(
        Text, 
        nullable=False, 
        primary_key=True, 
        default=text('(lower(hex(randomblob(4))))'), 
        server_default=text('(lower(hex(randomblob(4))))')
    )
    
    body = Column(JSON, nullable=False)
    date_added = Column(DATETIME, nullable=False)

    def __repr__(self):
        return f"<{self.__class__.__name__}{self.__dict__}>"

    @declared_attr
    def __tablename__(cls):
        return re.sub(r'(?<!^)(?=[A-Z])', '_', cls.__name__).lower()

class JsonSchema(Base, DocumentMetadata):
    title = Column(Text, nullable=False)

SQLAlchemy + SQLite 中的插入操作因 NULL 标识键错误而失败

Insert operation fails with NULL identity key Error in SQLAlchemy + SQLite

sqlite

sqlalchemy