SQLAlchemy ORMによるレコードの挿入/更新時にSQL関数を使う方法

この記事は公開から2年以上経過しています。

SQLAlchemy ORMでレコードの挿入や更新を行う際、生成されるInsert/Update文のパラメーターでSYSDATETIME()やCURRENT_TIMESTAMPのようなサーバー側で実行されるSQL関数を指定する方法。

対応

挿入/更新レコードで値を設定する際にSQLAlchemyのtext()でSQL関数を渡す。

検証用テーブル定義(SQLite)

CREATE TABLE TEST_TBL(
    ID INTEGER PRIMARY KEY AUTOINCREMENT,
    VAL TEXT,
    UPDATED_TIME TEXT NOT NULL
)

サンプルソースコード

import sqlalchemy as sqlalc
import sqlalchemy.ext.declarative as alcexdecl
import sqlalchemy.orm as alcorm
import os

Base = alcexdecl.declarative_base()

class TestTbl(Base):
    __tablename__ = "TEST_TBL"
    id = sqlalc.Column(sqlalc.Integer, primary_key=True)
    val = sqlalc.Column(sqlalc.String)
    updated_time = sqlalc.Column(sqlalc.DateTime)

db_file = '/var/tmp/test.db'

engine = sqlalc.create_engine(
    f'sqlite:///{db_file}', echo=True)

with alcorm.sessionmaker(engine)() as ssn:
    test_tbl_row = TestTbl(
        val='test',
        updated_time = sqlalc.text('CURRENT_TIMESTAMP'))
    ssn.add(test_tbl_row)
    ssn.commit()

生成SQL(ログ):

INSERT INTO "TEST_TBL" (val, updated_time) VALUES (?, CURRENT_TIMESTAMP)

実行SQLのパラメーターにSQL関数CURRENT_TIMESTAMPが使用されているのが確認できます。

以上です。