SQLAlchemy SQLite中的插入操作因NULL標識鍵錯誤而失敗

tldr; 在嘗試將記錄插入到json_schema使用 ORM 映射器類呼叫的表中后，我收到了錯誤：sqlalchemy.orm.exc.FlushError: Instance <JsonSchema at 0x10657f730> has a NULL identity key. 該錯誤的一部分內容為“確保此 flush() 不會在不適當的時間發生，例如在 load() 事件中。” ，并且我不確定“不適當的時間”在這種情況下是什么意思，或者它如何進一步幫助我解決問題。

在一個 sqlite 資料庫中，我有一個名為的表json_schema，看起來像這樣。此表中的主鍵是一個 4 位元組的十六進制文本字串，由(lower(hex(randomblob(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.

我檢查了這個錯誤中提到的第一個問題—— “檢查資料庫表是否允許生成新的主鍵值” ——通過測驗插入INSERT（id未指定）。這有效，所以不是錯誤的來源。

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

接下來，我檢查了“映射的 Column 物件是否配置為期望這些生成的值”。在 ORM 類中。在下面的代碼片段中，您可以看到繼承的id列JsonSchema確實具有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()發生在“不適當的時間”？

uj5u.com熱心網友回復：

SQLAlchemy 還不支持 SQLite 的 RETURNING，在這種情況下，主鍵不由AUTOINCREMENT. 在這種情況下，SQLAlchemy 必須自己執行默認生成函式并顯式插入生成的值。

要實作這一點，要么：

1. 在列定義中更改server_default=text('(lower(hex(randomblob(4))))')為default=text('(lower(hex(randomblob(4))))')

   • 表定義將不再包含該DEFAULT子句

2. 添加default=text('(lower(hex(randomblob(4))))')到列定義，留server_default在原地

   • 該表將保留該DEFAULT子句，盡管 SQLAlchemy 將始終覆寫它。

這記錄在Fetching Server-Generated Defaults中，特別是在案例 4：不支持主鍵、回傳或等效部分中。

第二種方法的新 ORM-mapper 類：

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

標籤：sqlite sqlalchemy

上一篇：如何以更有效的方式撰寫此代碼？

下一篇：如果不存在則插入，否則回傳值