SQLAlchemyとSQLServerの組み合わせでupdlock, rowlock等をかける方法

1 min read読了の目安(約1400字

TL;DR

session.query(Table).with_hint("WITH (ROWLOCK)").where(Table.count > 0)
# SELECT * FROM Table WITH (ROWLOCK) WHERE count > 0

やりたいこと

SQLAlchemyではロックをかける場合普通ならwith_for_updateメソッドを用いますが、以下の例を見ての通り、SQLServerにFOR UPDATEはないのでこの記述は無視されます。

from sqlalchemy.sql import select, table, literal_column
from sqlalchemy.dialects import mssql

tbl = table("MyTable")
stmt = select([tbl]).where(literal_column("hoge") > 0).with_for_update()

print(stmt.compile(dialect=mssql.dialect()))
# SELECT
# FROM [MyTable]
# WHERE hoge > :hoge_1

解決策

SQLServerではロックはFROM句の末尾にロックヒントという形でWITH (ROWLOCK, UDPLOCK)などと記述します。SQLAlchemyではwith_hintというメソッドを用いて以下のようにヒント文字列を埋め込むことができます。

from sqlalchemy.sql import select, table, literal_column
from sqlalchemy.dialects import mssql

tbl = table("MyTable")
stmt = select([tbl]).with_hint(tbl, "WITH (ROWLOCK)").where(literal_column("count") > 0)

print(stmt.compile(dialect=mssql.dialect()))
# SELECT
# FROM [MyTable] WITH (ROWLOCK)
# WHERE count > :count_1

めでたしめでたし

参考

https://docs.sqlalchemy.org/en/14/core/selectable.html#sqlalchemy.sql.expression.Select.with_hint

https://docs.sqlalchemy.org/en/14/core/selectable.html#sqlalchemy.sql.expression.Select.with_for_update

https://stackoverflow.com/questions/56030825/sqlalchemy-query-api-not-working-correctly-with-hints

https://www.m3tech.blog/entry/sqlalchemy-tutorial