From 09cf4faf124c9955cbed1bcb4241f13ac47f2e5c Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sun, 1 Nov 2026 12:58:11 -0700 Subject: [PATCH 1/2] feat: add lock(value) to the query builder for SKIP LOCKED claims Why: - Concurrent workers claiming pending rows waited on each other under FOR UPDATE. - The builder could not render FOR UPDATE SKIP LOCKED, so claims fell back to raw SQL. How: - QueryBuilder.lock(value=True) follows Laravel: True is the update lock, False the shared lock, a string is the dialect clause rendered verbatim. - lock_for_update() and shared_lock() delegate to it; Model.lock proxies it. - process_locks() renders unknown lock strings verbatim; SQLite renders no lock. - Grammar snapshots on four dialects and a Postgres test where concurrent fetch-then-update claims take disjoint rows. - ADR 063 and its index row. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VCMGYMYVHdzHAFaDeRWDyD --- docs/adr/063-lock-with-a-value.md | 35 ++++++++ docs/adr/AGENTS.md | 1 + .../masoniteorm/models/builder.py | 15 ++-- .../masoniteorm/models/model.py | 4 + .../masoniteorm/query/grammars/BaseGrammar.py | 2 +- .../query/grammars/SQLiteGrammar.py | 3 + .../postgres/test_postgres_skip_locked.py | 85 +++++++++++++++++++ .../query/grammars/test_lock_value_grammar.py | 85 +++++++++++++++++++ 8 files changed, 224 insertions(+), 6 deletions(-) create mode 100644 docs/adr/063-lock-with-a-value.md create mode 100644 fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py create mode 100644 fastapi_startkit/tests/masoniteorm/query/grammars/test_lock_value_grammar.py diff --git a/docs/adr/063-lock-with-a-value.md b/docs/adr/063-lock-with-a-value.md new file mode 100644 index 00000000..82b15f06 --- /dev/null +++ b/docs/adr/063-lock-with-a-value.md @@ -0,0 +1,35 @@ +# 063. `lock(value)` on the query builder, for `SKIP LOCKED` claims + +Date: 2026-11-01 + +## Problem + +A worker that claims pending rows from a shared table, such as a job queue or a per-user work list, needs concurrent workers to take different rows without waiting on each other. ADR 061 added `lock_for_update()` and `shared_lock()`, but a second worker that asks for the same rows with `FOR UPDATE` waits for the first to commit. Postgres 9.5+, MySQL 8.0.1+ and MariaDB 10.6+ solve this with `FOR UPDATE SKIP LOCKED`, and SQL Server with the `READPAST` hint. The builder could not render either, so applications wrote the claim as raw SQL. + +## Alternatives + +1. **A `skip_locked` flag: `lock_for_update(skip_locked=True)`.** It adds a new grammar key per option and grows with `NOWAIT`, `FOR NO KEY UPDATE` and the rest. Laravel has no such flag. +2. **`UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED) RETURNING ...` in one statement.** It needs `RETURNING` support in every grammar and hides the claim inside a single statement. +3. **Laravel's `lock($value = true)`.** `true` takes the update lock, `false` the shared lock, and a string is the dialect's lock clause, rendered verbatim. Laravel's database queue claims a job this way: `lock('FOR UPDATE SKIP LOCKED')->orderBy('id')->first()` in a transaction, then an update of that job by id. + +## Decision + +Option 3. + +- `QueryBuilder.lock(value: bool | str = True)` sets `_lock` to `"update"` for `True`, `"share"` for `False`, or the string itself. `lock_for_update()` is `lock(True)` and `shared_lock()` is `lock(False)`. The last call wins. `Model.lock(value)` proxies to a new query. +- `BaseGrammar.process_locks()` renders the dialect clause for `"update"` and `"share"` and any other string verbatim, in the same slot: after `LIMIT`/`OFFSET` on Postgres and MySQL, and after the table on MSSQL (`WITH(ROWLOCK, UPDLOCK, READPAST)`). `SQLiteGrammar.process_locks()` renders nothing, as it already did for both locks. SQLite locks the whole database for a write transaction, so the same code runs in tests. +- A locked builder used as a `where_in` subquery keeps its lock, because the subquery compiles through the sub-builder's own grammar. +- A claim reads the matching rows with a lock inside `connection.transaction()`, iterates them and updates each by key, so the caller knows which rows it claimed without `RETURNING`. + +ADR 061 rejected a stringly `lock(mode)` because an unknown mode silently rendered no lock. A string now renders verbatim, so a typo fails in the database instead of running unlocked. + +A lock string is SQL, not a binding. It must be a literal written in the application's code, such as a module constant, and never built from request input, configuration a user can change, or any other untrusted value. `lock()` does not escape or validate it, just as `select_raw` does not, because the set of valid clauses differs by dialect and version. + +The caller picks the clause for its dialect, as in Laravel's `DatabaseQueue::getLockForPopping()`. A string meant for one dialect is invalid SQL on another, except on SQLite, where every lock is a no-op. + +## Validation + +- `tests/masoniteorm/query/grammars/test_lock_value_grammar.py` snapshots `to_qmark()` on SQLite, MySQL, Postgres and MSSQL: `lock()` and `lock(True)` take the update lock, `lock(False)` the shared lock, a string renders verbatim after `ORDER BY` and `LIMIT` (after the table and with `TOP` on MSSQL, nothing on SQLite), a locked `where_in` subquery keeps its lock, and the last call wins. +- `tests/masoniteorm/postgres/test_postgres_skip_locked.py` runs concurrent transactions on Postgres. Three workers each read up to two pending rows with `lock("FOR UPDATE SKIP LOCKED")`, update each by id and hold their transaction. They claim 2, 2 and 1 of five rows, no row twice, and all five end up processing. A worker using a plain `lock()` waits for the holder and never gets the held row. +- `test_lock_grammar.py`, `test_sqlite_lock.py` and `test_postgres_lock.py` from ADR 061 pass unchanged. +- The full framework suite passes with coverage at or above the threshold. diff --git a/docs/adr/AGENTS.md b/docs/adr/AGENTS.md index 22271811..60d59061 100644 --- a/docs/adr/AGENTS.md +++ b/docs/adr/AGENTS.md @@ -51,3 +51,4 @@ Read this index first, then open the records relevant to the change. Before impl | [059](059-subquery-where-in-and-select-exists.md) | 2026-10-31 | Subqueries in `where_in`/`where_not_in`, `where_not_exists` and `add_select_exists` | Membership, anti-join and per-row exists flags needed `DB.select`: `where_in(col, builder)` raised `'QueryBuilder' object is not iterable` and no method emitted `NOT EXISTS` or an `EXISTS` column. `where_in`/`where_not_in` now take a builder or a callable (like `select_sub`) and render `col IN (subquery)` via `SubSelectExpression`; iterables are unchanged. `where_exists`/`or_where_exists` also take a callable; new `where_not_exists`/`or_where_not_exists`. `add_select_exists(alias, query)` appends `EXISTS(subquery) AS alias` (selecting `{table}.*` first when nothing is selected) through a `SubGroupExpression(exists=True)` and the grammar's `select_exists_string()`; MSSQL renders `CASE WHEN EXISTS(...) THEN 1 ELSE 0 END`. A shared `_resolve_subquery` raises one `TypeError`. Subquery bindings stay bound and land in SQL order. `Model` proxies all. Limits: Postgres returns `bool`, others `1`/`0`; `NOT IN` over a subquery with `NULL` matches nothing. Chosen over raw SQL or separate `*_sub` method names. | | [061](061-builder-row-locks.md) | 2026-10-31 | Row locks on the query builder: `lock_for_update` and `shared_lock` | The grammars already rendered a per-dialect lock clause, but no builder method set it and `get_grammar()` never passed `lock`, so apps wrote `SELECT ... FOR UPDATE` by hand. `QueryBuilder.lock_for_update()`/`shared_lock()` set `_lock` to `"update"`/`"share"` (last call wins) and `get_grammar()` passes it: Postgres `FOR UPDATE`/`FOR SHARE`, MySQL `FOR UPDATE`/`LOCK IN SHARE MODE`, SQLite a no-op, and MSSQL now `WITH(ROWLOCK, UPDLOCK, HOLDLOCK)`/`WITH(ROWLOCK, HOLDLOCK)` instead of a bare `ROWLOCK` hint. `Model` proxies both. Only selects lock; `new()` groups and the `paginate()` count start unlocked. Chosen over update-first, whose `FOR NO KEY UPDATE` does not wait for an in-flight foreign-key insert, and over a stringly `lock(mode)`. Known limit: Postgres refuses a lock with aggregates, `DISTINCT` or `GROUP BY`. | | [062](062-update-uses-the-updated-table.md) | 2026-11-01 | `update()` compiles against the updated table | `update()` built its grammar with `self.grammar()`, keeping the default table `users`, so MySQL and MSSQL prefixed every `SET`/`WHERE` column with `users` whatever table was updated (`UPDATE [queues] SET [users].[status] = ?`). It now passes `table=self._table`. Postgres and SQLite do not prefix update columns. `update()` still returns the row count; a queue claim reads locked rows, iterates and updates each by key, as Laravel's database queue does, so no `UPDATE ... RETURNING` is added. | +| [063](063-lock-with-a-value.md) | 2026-11-01 | `lock(value)` on the query builder, for `SKIP LOCKED` claims | Concurrent workers claiming pending rows waited on each other under `FOR UPDATE`, and the builder could not render `FOR UPDATE SKIP LOCKED`. Laravel's `QueryBuilder.lock(value=True)`: `True` takes the update lock, `False` the shared lock, a string is a trusted, code-literal dialect clause rendered verbatim (never user input) (e.g. `FOR UPDATE SKIP LOCKED`, MSSQL `WITH(ROWLOCK, UPDLOCK, READPAST)`); SQLite renders nothing. `lock_for_update()`/`shared_lock()` delegate to it and `Model.lock` proxies it. A locked `where_in` subquery keeps its lock. A claim reads locked rows in a transaction, iterates them and updates each by key, as Laravel's database queue does. Chosen over a `skip_locked` flag and over `UPDATE ... RETURNING`. | diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/builder.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/builder.py index c48d86f2..6672e480 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/builder.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/builder.py @@ -91,7 +91,7 @@ def __init__(self, connection: "Connection", grammar: Any, processor: Any): self._group_by = () self._having = () self._distinct = False - self._lock: Literal["update", "share", False] = False + self._lock: str | Literal[False] = False self._sql = "" self._bindings = () @@ -393,13 +393,18 @@ def distinct(self) -> "Self": self._distinct = True return self - def lock_for_update(self) -> "Self": - self._lock = "update" + def lock(self, value: bool | str = True) -> "Self": + if isinstance(value, str): + self._lock = value + else: + self._lock = "update" if value else "share" return self + def lock_for_update(self) -> "Self": + return self.lock(True) + def shared_lock(self) -> "Self": - self._lock = "share" - return self + return self.lock(False) async def aggregate(self, function: str, column: str): self._aggregates += (AggregateExpression(function, column),) diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/model.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/model.py index d2770037..c989bc57 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/model.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/models/model.py @@ -243,6 +243,10 @@ def not_between(cls, column: str, low: Any, high: Any) -> "QueryBuilder[Self]": def distinct(cls) -> "QueryBuilder[Self]": return cls.query().distinct() + @classmethod + def lock(cls, value: bool | str = True) -> "QueryBuilder[Self]": + return cls.query().lock(value) + @classmethod def lock_for_update(cls) -> "QueryBuilder[Self]": return cls.query().lock_for_update() diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py index cab3e71c..07d70531 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py @@ -567,7 +567,7 @@ def process_offset(self): return self.offset_string().format(offset=self._offset, limit=self._limit or 1) def process_locks(self): - return self.locks.get(self.lock, "") + return self.locks.get(self.lock, self.lock or "") def process_having(self, qmark=False): """Compiles having expression. diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/SQLiteGrammar.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/SQLiteGrammar.py index e5379c4a..04f93e3f 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/SQLiteGrammar.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/SQLiteGrammar.py @@ -37,6 +37,9 @@ class SQLiteGrammar(BaseGrammar): locks = {"share": "", "update": ""} + def process_locks(self): + return "" + def select_format(self): return "SELECT {keyword} {columns} FROM {table} {joins} {wheres} {group_by} {having} {order_by} {limit} {offset} {lock}" diff --git a/fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py b/fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py new file mode 100644 index 00000000..529f0ac2 --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py @@ -0,0 +1,85 @@ +import asyncio +import contextlib + +from fastapi_startkit.masoniteorm import Model + +from .fixtures.db import DB, schema +from .test_case import TestCase + +BLOCKED_FOR = 0.3 +SETTLES_WITHIN = 5 + + +class PskQueueItem(Model): + __table__ = "psk_queue_items" + __timestamps__ = None + id: int + status: str + + +def pending(batch): + return PskQueueItem.where("status", "pending").order_by("id").limit(batch) + + +class TestPostgresSkipLocked(TestCase): + async def asyncSetUp(self): + await super().asyncSetUp() + self.release = asyncio.Event() + self.holders = [] + await schema.drop_table_if_exists("psk_queue_items") + async with await schema.create("psk_queue_items") as table: + table.increments("id") + table.string("status") + await PskQueueItem.query().insert([{"status": "pending"} for _ in range(5)]) + + async def asyncTearDown(self): + self.release.set() + for task in self.holders: + with contextlib.suppress(Exception, asyncio.CancelledError): + await asyncio.wait_for(task, SETTLES_WITHIN) + await schema.drop_table_if_exists("psk_queue_items") + await super().asyncTearDown() + + async def claim_and_hold(self, batch, lock): + claimed = asyncio.get_running_loop().create_future() + + async def worker(): + async with DB.connection().transaction(): + ids = [] + for item in await lock(pending(batch)).get(): + await PskQueueItem.where("id", item.id).update({"status": "processing"}) + ids.append(item.id) + claimed.set_result(ids) + await self.release.wait() + + self.holders.append(asyncio.ensure_future(worker())) + return claimed + + async def test_concurrent_claims_skip_each_others_rows(self): + def skip_locked(query): + return query.lock("FOR UPDATE SKIP LOCKED") + + first = await asyncio.wait_for(await self.claim_and_hold(2, skip_locked), SETTLES_WITHIN) + second = await asyncio.wait_for(await self.claim_and_hold(2, skip_locked), SETTLES_WITHIN) + third = await asyncio.wait_for(await self.claim_and_hold(2, skip_locked), SETTLES_WITHIN) + + self.assertEqual(len(first), 2) + self.assertEqual(len(second), 2) + self.assertEqual(len(third), 1) + self.assertEqual(len(set(first) | set(second) | set(third)), 5) + + self.release.set() + await asyncio.gather(*self.holders) + self.assertEqual(await PskQueueItem.where("status", "processing").count(), 5) + + async def test_a_plain_update_lock_waits_for_the_holder(self): + def skip_locked(query): + return query.lock("FOR UPDATE SKIP LOCKED") + + held = await asyncio.wait_for(await self.claim_and_hold(1, skip_locked), SETTLES_WITHIN) + blocked = await self.claim_and_hold(1, lambda query: query.lock()) + + done, _ = await asyncio.wait({blocked}, timeout=BLOCKED_FOR) + self.assertEqual(done, set()) + self.release.set() + self.assertNotIn(held[0], await asyncio.wait_for(blocked, SETTLES_WITHIN)) diff --git a/fastapi_startkit/tests/masoniteorm/query/grammars/test_lock_value_grammar.py b/fastapi_startkit/tests/masoniteorm/query/grammars/test_lock_value_grammar.py new file mode 100644 index 00000000..fbd8e14a --- /dev/null +++ b/fastapi_startkit/tests/masoniteorm/query/grammars/test_lock_value_grammar.py @@ -0,0 +1,85 @@ +import unittest + +from fastapi_startkit.masoniteorm.models.builder import QueryBuilder +from fastapi_startkit.masoniteorm.query.grammars.MSSQLGrammar import MSSQLGrammar +from fastapi_startkit.masoniteorm.query.grammars.MySQLGrammar import MySQLGrammar +from fastapi_startkit.masoniteorm.query.grammars.PostgresGrammar import PostgresGrammar +from fastapi_startkit.masoniteorm.query.grammars.SQLiteGrammar import SQLiteGrammar + +GRAMMARS = { + "sqlite": SQLiteGrammar, + "mysql": MySQLGrammar, + "postgres": PostgresGrammar, + "mssql": MSSQLGrammar, +} + +SKIP_LOCKED = { + "sqlite": "FOR UPDATE SKIP LOCKED", + "mysql": "FOR UPDATE SKIP LOCKED", + "postgres": "FOR UPDATE SKIP LOCKED", + "mssql": "WITH(ROWLOCK, UPDLOCK, READPAST)", +} + +EXPECTED = { + "sqlite": { + "update": 'SELECT * FROM "queues" WHERE "queues"."id" = ?', + "share": 'SELECT * FROM "queues" WHERE "queues"."id" = ?', + "string": 'SELECT * FROM "queues" WHERE "queues"."status" = ? ORDER BY "id" ASC LIMIT 1', + "subquery": 'SELECT * FROM "queues" WHERE "queues"."id" IN (SELECT "queues"."id" FROM "queues" WHERE "queues"."status" = ? LIMIT 1)', + }, + "mysql": { + "update": "SELECT * FROM `queues` WHERE `queues`.`id` = ? FOR UPDATE", + "share": "SELECT * FROM `queues` WHERE `queues`.`id` = ? LOCK IN SHARE MODE", + "string": "SELECT * FROM `queues` WHERE `queues`.`status` = ? ORDER BY `id` ASC LIMIT 1 FOR UPDATE SKIP LOCKED", + "subquery": "SELECT * FROM `queues` WHERE `queues`.`id` IN (SELECT `queues`.`id` FROM `queues` WHERE `queues`.`status` = ? LIMIT 1 FOR UPDATE SKIP LOCKED)", + }, + "postgres": { + "update": 'SELECT * FROM "queues" WHERE "queues"."id" = ? FOR UPDATE', + "share": 'SELECT * FROM "queues" WHERE "queues"."id" = ? FOR SHARE', + "string": 'SELECT * FROM "queues" WHERE "queues"."status" = ? ORDER BY "id" ASC LIMIT 1 FOR UPDATE SKIP LOCKED', + "subquery": 'SELECT * FROM "queues" WHERE "queues"."id" IN (SELECT "queues"."id" FROM "queues" WHERE "queues"."status" = ? LIMIT 1 FOR UPDATE SKIP LOCKED)', + }, + "mssql": { + "update": "SELECT * FROM [queues] WITH(ROWLOCK, UPDLOCK, HOLDLOCK) WHERE [queues].[id] = ?", + "share": "SELECT * FROM [queues] WITH(ROWLOCK, HOLDLOCK) WHERE [queues].[id] = ?", + "string": "SELECT TOP 1 * FROM [queues] WITH(ROWLOCK, UPDLOCK, READPAST) WHERE [queues].[status] = ? ORDER BY [id] ASC", + "subquery": "SELECT * FROM [queues] WHERE [queues].[id] IN (SELECT TOP 1 [queues].[id] FROM [queues] WITH(ROWLOCK, UPDLOCK, READPAST) WHERE [queues].[status] = ?)", + }, +} + + +def qb(grammar): + return QueryBuilder(connection=None, grammar=grammar, processor=None).table("queues") + + +class TestLockValueGrammar(unittest.TestCase): + def test_lock_true_takes_the_update_lock(self): + for name, grammar in GRAMMARS.items(): + with self.subTest(grammar=name): + self.assertEqual(qb(grammar).where("id", 7).lock().to_qmark(), EXPECTED[name]["update"]) + self.assertEqual(qb(grammar).where("id", 7).lock(True).to_qmark(), EXPECTED[name]["update"]) + + def test_lock_false_takes_the_shared_lock(self): + for name, grammar in GRAMMARS.items(): + with self.subTest(grammar=name): + self.assertEqual(qb(grammar).where("id", 7).lock(False).to_qmark(), EXPECTED[name]["share"]) + + def test_a_lock_string_renders_verbatim_except_on_sqlite(self): + for name, grammar in GRAMMARS.items(): + with self.subTest(grammar=name): + query = qb(grammar).where("status", "pending").order_by("id").limit(1).lock(SKIP_LOCKED[name]) + self.assertEqual(query.to_qmark(), EXPECTED[name]["string"]) + self.assertEqual(list(query._bindings), ["pending"]) + + def test_a_locked_subquery_keeps_its_lock(self): + for name, grammar in GRAMMARS.items(): + with self.subTest(grammar=name): + claimable = qb(grammar).select("id").where("status", "pending").limit(1).lock(SKIP_LOCKED[name]) + query = qb(grammar).where_in("id", claimable) + self.assertEqual(query.to_qmark(), EXPECTED[name]["subquery"]) + + def test_the_last_lock_call_wins(self): + for name, grammar in GRAMMARS.items(): + with self.subTest(grammar=name): + query = qb(grammar).where("id", 7).lock("FOR UPDATE NOWAIT").lock_for_update() + self.assertEqual(query.to_qmark(), EXPECTED[name]["update"]) From ec3f0f182ac7729d57a335a9c775959feafb741e Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sun, 1 Nov 2026 13:47:14 -0700 Subject: [PATCH 2/2] test: claim a free row with one update through a SKIP LOCKED subquery Why: - A claim can also be a single builder-built UPDATE ... WHERE id = (SELECT ... FOR UPDATE SKIP LOCKED), with no RETURNING and no raw SQL. - _compile_update kept a docstring after #345 touched it, against the no-docstring rule. How: - A Postgres test holds one row in another transaction and asserts that update() through a locked where subquery claims exactly one other row; with a plain lock() subquery the update blocks and the test fails. - Remove the _compile_update docstring. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VCMGYLLSTz3VTHCp6XygtL --- .../masoniteorm/query/grammars/BaseGrammar.py | 12 ------------ .../postgres/test_postgres_skip_locked.py | 15 +++++++++++++++ 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py b/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py index 07d70531..e5858b53 100644 --- a/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py +++ b/fastapi_startkit/src/fastapi_startkit/masoniteorm/query/grammars/BaseGrammar.py @@ -157,18 +157,6 @@ def _compile_select(self, qmark=False): return self def _compile_update(self, query=None, values=None, qmark=False): - """Compiles an update query statement. - - Arguments: - query -- The QueryBuilder instance (provides table/wheres context). - values {list} -- Runtime UpdateQueryExpression list. - - Keyword Arguments: - qmark {bool} -- Use positional ? bindings. (default: {False}) - - Returns: - self - """ self._action = "update" if values is not None: diff --git a/fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py b/fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py index 529f0ac2..f9c3904c 100644 --- a/fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py +++ b/fastapi_startkit/tests/masoniteorm/postgres/test_postgres_skip_locked.py @@ -83,3 +83,18 @@ def skip_locked(query): self.assertEqual(done, set()) self.release.set() self.assertNotIn(held[0], await asyncio.wait_for(blocked, SETTLES_WITHIN)) + + async def test_one_update_through_a_skip_locked_subquery_claims_a_free_row(self): + held = await asyncio.wait_for( + await self.claim_and_hold(1, lambda query: query.lock("FOR UPDATE SKIP LOCKED")), SETTLES_WITHIN + ) + next_free = pending(1).select("id").lock("FOR UPDATE SKIP LOCKED") + + updated = await asyncio.wait_for( + PskQueueItem.where("id", next_free).update({"status": "processing"}), SETTLES_WITHIN + ) + + processing = [item.id for item in await PskQueueItem.where("status", "processing").get()] + self.assertEqual(updated, 1) + self.assertEqual(len(processing), 1) + self.assertNotIn(held[0], processing)