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

하고 싶은 일

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