Метод with_for_update класса Query
Метод with_for_update класса Query
добавляет к SQL-запросу конструкцию
FOR UPDATE, которая блокирует выбранные
строки до завершения текущей транзакции.
Это предотвращает изменение этих строк
другими транзакциями и защищает от
состояния гонки. Первым параметром можно
передать булево значение read, вторым -
nowait, третьим - of,
четвертым - skip_locked.
Синтаксис
Query.with_for_update(read=False, nowait=False, of=None, skip_locked=False)
Пример
Давайте создадим модель и получим строку с блокировкой:
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(50))
text: Mapped[str] = mapped_column(String(200))
status: Mapped[str] = mapped_column(String(20))
num: Mapped[int]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add(Article(title='article', text='text', status='new', num=1))
session.commit()
res = session.query(Article).with_for_update().first()
print(res.title)
Результат выполнения кода:
"article"
Итоговый SQL запрос:
SELECT articles.id AS articles_id, articles.title AS articles_title, articles.text AS articles_text, articles.status AS articles_status, articles.num AS articles_num
FROM articles
FOR UPDATE
Пример
Давайте применим блокировку с параметром
nowait, чтобы не ждать освобождения
строки:
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(50))
text: Mapped[str] = mapped_column(String(200))
status: Mapped[str] = mapped_column(String(20))
num: Mapped[int]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add(Article(title='article', text='text', status='new', num=1))
session.commit()
res = session.query(Article).with_for_update(nowait=True).all()
print(len(res))
Результат выполнения кода:
"1"
Итоговый SQL запрос:
SELECT articles.id AS articles_id, articles.title AS articles_title, articles.text AS articles_text, articles.status AS articles_status, articles.num AS articles_num
FROM articles
FOR UPDATE NOWAIT
Пример
Давайте передадим параметр skip_locked,
чтобы пропустить уже заблокированные строки:
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class Base(DeclarativeBase):
pass
class Article(Base):
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(String(50))
text: Mapped[str] = mapped_column(String(200))
status: Mapped[str] = mapped_column(String(20))
num: Mapped[int]
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
with Session(engine) as session:
session.add(Article(title='article', text='text', status='new', num=1))
session.commit()
res = session.query(Article).with_for_update(skip_locked=True).all()
print(res[0].title)
Результат выполнения кода:
"article"
Итоговый SQL запрос:
SELECT articles.id AS articles_id, articles.title AS articles_title, articles.text AS articles_text, articles.status AS articles_status, articles.num AS articles_num
FROM articles
FOR UPDATE SKIP LOCKED
