SQLAlchemy와 SQL 서버의 조합으로 updlock,rowlock 등을 추가하는 방법

TL;DR

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

하고 싶은 일

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

해결책

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

참고 자료