From a4a3feb2dc6eee8287b3c027aed417695aeabe75 Mon Sep 17 00:00:00 2001 From: Alexey Shalaev <75322386+AlexeyShalaev@users.noreply.github.com> Date: Fri, 16 Oct 2026 18:49:48 +0400 Subject: [PATCH] fix!: run query() in a READ ONLY transaction, as documented BREAKING CHANGE: uow.query() now opens its transaction READ ONLY. A write inside it (INSERT, UPDATE, DELETE, DDL, nextval()) fails with SQLSTATE 25006 instead of being rolled back silently, or persisted when followed by an explicit session.commit(). Move writes to uow.transaction(). --- docs/agents.md | 25 ++-- docs/guide/advanced.md | 17 ++- sqlalchemy_foundation_kit/uow/protocols.py | 4 +- sqlalchemy_foundation_kit/uow/sqlalchemy.py | 46 ++++-- tests/integration/test_uow_integration.py | 151 +++++++++++++++++++- tests/unit/contrib/telemetry/test_uow.py | 1 + 6 files changed, 217 insertions(+), 27 deletions(-) diff --git a/docs/agents.md b/docs/agents.md index af6b65a..4f9a3fc 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -61,8 +61,8 @@ Five nouns, in the order a request passes through them. application is correct and intended. * **A context manager on it opens exactly one session** and closes it on exit. `transaction()` commits on success and rolls back on exception; `managed_session()` - hands you the session and commits nothing; `query()` starts no transaction of its own - and throws away anything you write. + hands you the session and commits nothing; `query()` runs one `READ ONLY` transaction + that is rolled back on exit, so a write inside it raises. * **The transaction object** is what the context manager yields: your subclass of `AsyncSQLAlchemyUowTransaction`, constructed with that session, exposing your repositories as properties. `tx.session` is the `AsyncSession` underneath; @@ -127,7 +127,7 @@ async def main() -> None: async with uow.transaction() as tx: # commits on exit, rolls back on exception await tx.users.add("user@example.com") - async with uow.query() as qx: # read only: writes here are discarded + async with uow.query() as qx: # READ ONLY: a write here raises (SQLSTATE 25006) users = (await qx.session.execute(select(UserDB))).scalars().all() await manager.aclose() @@ -235,7 +235,7 @@ a wait histogram that never moves. |---|---|---|---|---| | `transaction(isolation_level=None, flush_before_commit=None)` | `session.begin()` | flush, then **commit** | rollback | `T` | | `managed_session(isolation_level=None)` | `await session.begin()` | nothing — **you** commit | rollback | `(T, AsyncSession)` | -| `query(isolation_level=None)` | none started | nothing; session closes | session closes | `T` | +| `query(isolation_level=None)` | one `READ ONLY` transaction | rolled back; session closes | rolled back; session closes | `T` | | `open_session(isolation_level=None)` | none | nothing | nothing | `AsyncSession` | `open_session` is the documented extension point: override it in a subclass to set a GUC, @@ -391,9 +391,13 @@ decorator. 4. **The commit belongs to the unit of work, never to a repository.** A repository `flush()`es; `transaction()` commits. A `session.commit()` inside a repository ends the transaction under everything else that was going to share it. -5. **`query()` throws writes away.** It starts no transaction and commits nothing, so the - session's implicit transaction is rolled back at close. An `INSERT` issued there is - silently gone — verified, not theorised. Use `transaction()` for anything that writes. +5. **`query()` is a read-only transaction.** It opens one `READ ONLY` transaction and rolls + it back on exit, so an `INSERT`, `UPDATE`, `DELETE`, DDL or `nextval()` inside it fails + with SQLSTATE 25006. Reads, `SET LOCAL` and advisory locks work. It is still a + transaction: the locks it takes (`AccessShareLock` on every table read) are held until + the block exits, `now()` is fixed at its first statement, and a slow `await` inside it + leaves the connection `idle in transaction`. Keep it short; use `transaction()` for + anything that writes. On 0.4.0 and earlier it was not read-only — see the table below. 6. **One session, one task.** The session a context manager yields is not concurrency-safe: `asyncio.gather` over statements on the same `tx` raises `IllegalStateChangeError`. Fan out by opening a `transaction()` per task, not by sharing @@ -414,8 +418,8 @@ decorator. new replicas holding different locks for the same name until it finishes. 9. **An advisory lock is only held while its transaction is.** `pg_try_advisory_xact_lock` releases at transaction end, so take it inside `transaction()` or `managed_session()` - and do the protected work in the same block. Taking one in `query()` is a no-op with a - `True` return. + and do the protected work in the same block. One taken in `query()` is held until the + block exits, then released — it protects only what that read-only block does. 10. **`managed_session()` commits nothing.** Leaving the block without calling `await session.commit()` rolls everything back at close, quietly. Use `transaction()` unless the commit decision genuinely depends on something you learn after writing. @@ -484,6 +488,7 @@ is what to do on the version the row names. | `postgres_db_pool_overflow`, `postgres_db_pool_checked_out` (0.4.0 and earlier) | sampled from the `checkout` / `checkin` events: `overflow` exported `QueuePool.overflow()`, which starts at `-pool_size` (`-9` with one connection open of 10), and since `checkin` fires before the connection is back, an idle pool kept reporting its last burst as checked out | read `pool.checkedout()` and `max(pool.overflow(), 0)` yourself | | `postgres_db_connection_timeouts_total` (0.4.0) | counted asyncpg statement timeouts — a slow query, on a healthy connection — and missed connect timeouts, which never reach `handle_error`. A database refusing connections moved no series at all | alert on `connection_errors_total{error_type="TimeoutError"}` for slow statements, and on the checkout wait for connections | | `uow.transaction()` with a failing flush (0.4.0 and earlier, `flush_before_commit=True`, the default) | logged `str(exc)` at WARNING as `extra={"error": ...}` before re-raising: the statement's parameters and PostgreSQL's `DETAIL: Key (...)=(...)` row values went to the logs, and `hide_parameters=True` removed only the parameters | `flush_before_commit=False`; the same exception still reaches the caller | +| `uow.query()` (0.4.0 and earlier) | documented as read-only and transaction-free, but ran an ordinary autobegun transaction: a write followed by `qx.session.commit()` was persisted, and without the commit it was rolled back silently | do reads in `query()` and every write in `transaction()` | | `postgres_db_connection_checkout_duration_seconds` (0.3.0 and earlier) | the only checkout histogram there was, and it measures the time between the `checkout` and `checkin` events — how long a connection was *held*, which is query duration seen from the pool. The time a caller waited for a connection, which is what the name suggests and what predicts a pool outage, was not exposed at all | none; the wait was not measurable from outside the pool | `IsolationLevel` itself was always fine — `READ_UNCOMMITTED`, `READ_COMMITTED`, @@ -512,7 +517,7 @@ class UserRepository: ``` ```python -# WRONG — query() starts no transaction, so this INSERT is rolled back at close +# WRONG — query() is a READ ONLY transaction: this INSERT raises (SQLSTATE 25006) async with uow.query() as qx: await qx.users.add("user@example.com") diff --git a/docs/guide/advanced.md b/docs/guide/advanced.md index ea7bcbc..f24c8e9 100644 --- a/docs/guide/advanced.md +++ b/docs/guide/advanced.md @@ -113,26 +113,33 @@ async with uow.transaction(isolation_level="READ COMMITTED") as tx: The level is set on the connection this block checks out, so it applies to this transaction and leaves the engine alone. It has to be set before the transaction starts, -which means the block's transaction opens as the context manager is entered — including -in `query()`, which otherwise waits for the first statement. To set a level for every +which means the block's transaction opens as the context manager is entered. To set a level for every session instead, pass `isolation_level=` to `AsyncSessionManager` or set `QuerySettings.isolation_level`. ### Read-Only Queries -For read-only operations without transaction overhead: +`query()` runs the block in one `READ ONLY` transaction and rolls it back on exit: ```python -# Query context - no transaction management async with uow.query() as qx: users = await qx.users.list_all(limit=100) - # No commit/rollback, just queries + # Rolled back on exit; nothing to commit # With isolation level async with uow.query(isolation_level=IsolationLevel.REPEATABLE_READ) as qx: users = await qx.users.list_all() ``` +PostgreSQL enforces the `READ ONLY`: an `INSERT`, `UPDATE`, `DELETE`, DDL statement or +`nextval()` inside the block fails with SQLSTATE 25006. Reads, `SET LOCAL` / +`set_config` and advisory locks work. Being a transaction, the block holds every lock it +takes — the `AccessShareLock` on each table it read, which makes an `ALTER TABLE` on that +table wait — until it exits, fixes `now()` at its first statement, and leaves the +connection `idle in transaction` across any `await` that is not a query. Behind a +transaction-mode PgBouncer it holds one server connection from its first statement to its +exit. Keep the block to the reads. + ### Manual Transaction Control For complex scenarios where you need explicit commit control: diff --git a/sqlalchemy_foundation_kit/uow/protocols.py b/sqlalchemy_foundation_kit/uow/protocols.py index 9ca86a8..8f14562 100644 --- a/sqlalchemy_foundation_kit/uow/protocols.py +++ b/sqlalchemy_foundation_kit/uow/protocols.py @@ -56,7 +56,7 @@ class AsyncUnitOfWork(Protocol, Generic[T_co]): Provides three modes of operation: - ``transaction()``: For write operations with automatic commit/rollback. - ``managed_session()``: For write operations with **manual** commit/rollback control. - - ``query()``: For read-only operations without transaction management. + - ``query()``: For reads, in one ``READ ONLY`` transaction that is rolled back on exit. """ def transaction( @@ -95,4 +95,4 @@ def managed_session( """ def query(self, isolation_level: str | None = None) -> AbstractAsyncContextManager[T_co]: - """Create a read-only query context without transaction management.""" + """Create a read-only query context: one ``READ ONLY`` transaction, rolled back on exit.""" diff --git a/sqlalchemy_foundation_kit/uow/sqlalchemy.py b/sqlalchemy_foundation_kit/uow/sqlalchemy.py index 06957c8..9eef1a1 100644 --- a/sqlalchemy_foundation_kit/uow/sqlalchemy.py +++ b/sqlalchemy_foundation_kit/uow/sqlalchemy.py @@ -7,6 +7,7 @@ from contextlib import asynccontextmanager from typing import Generic +from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from .._typing import T @@ -107,6 +108,23 @@ async def apply_isolation_level( await session.connection(execution_options={"isolation_level": normalized}) +async def _begin_read_only(session: AsyncSession) -> None: + """Make the session's transaction ``READ ONLY``. + + The ``postgresql_readonly`` execution option is applied when a connection is checked + out, so on a session that has not checked one out yet it goes with that checkout: + asyncpg then opens the transaction ``READ ONLY`` with no extra round trip, and the + dialect resets the flag when the connection returns to the pool. When ``open_session()`` + has already started the transaction -- an isolation level was applied, or an override + ran a statement -- ``SET TRANSACTION READ ONLY`` marks it instead; PostgreSQL accepts + that at any point in a transaction. + """ + if session.in_transaction(): + await session.execute(text("SET TRANSACTION READ ONLY")) + else: + await session.connection(execution_options={"postgresql_readonly": True}) + + @asynccontextmanager async def _owned_transaction(session: AsyncSession) -> AsyncIterator[None]: """Run a block inside a transaction, committing on success and rolling back on failure. @@ -227,7 +245,7 @@ class AsyncSQLAlchemyUnitOfWork(AsyncUnitOfWork[T], Generic[T]): Methods: transaction(): For write operations with automatic commit/rollback. - query(): For read-only operations without transaction management. + query(): For reads, in one READ ONLY transaction that is rolled back on exit. """ def __init__( @@ -432,11 +450,20 @@ async def query( self, isolation_level: IsolationLevel | str | None = None, ) -> AsyncIterator[T]: - """Create a read-only query context without transaction management. + """Create a read-only query context: one ``READ ONLY`` transaction, never committed. + + The block runs in a single transaction from its first statement until it exits, + when the transaction is rolled back. PostgreSQL enforces ``READ ONLY``: an + ``INSERT``, ``UPDATE``, ``DELETE`` or DDL statement, and functions such as + ``nextval()``, fail with SQLSTATE 25006. Reads, ``SET LOCAL`` / ``set_config`` and + advisory locks work. - This method is designed for read-only operations and does not start a transaction - or perform any commit/rollback. It's semantically clearer than managed_session() - for read operations. + It is still a transaction, with what that costs: every lock it takes -- the + ``AccessShareLock`` on each table read, an advisory xact lock -- is held until the + block exits, ``now()`` is fixed at the first statement, a slow ``await`` inside the + block leaves the connection ``idle in transaction``, and behind a transaction-mode + pooler the block holds one server connection from its first statement to exit. + Keep it short. Args: isolation_level: Optional transaction isolation level. @@ -451,13 +478,14 @@ async def query( async with uow.query() as qx: users = await qx.users.list_all() user = await qx.users.get_by_id(user_id) - # No commit/rollback - just closes session + # Rolled back on exit -- there is nothing to commit Note: - While this method is intended for read-only operations, SQLAlchemy does not enforce - this at the session level. It's up to the caller to ensure only read operations are performed. + Only the block's first transaction is read-only. Calling + ``session.commit()`` inside ``query()`` ends it, and the session starts an + ordinary one on its next statement. Write in :meth:`transaction` instead. """ async with self.open_session(isolation_level) as session: - # No transaction begin/commit - just yield the session + await _begin_read_only(session) uow = self._transaction_factory(session) yield uow diff --git a/tests/integration/test_uow_integration.py b/tests/integration/test_uow_integration.py index d06f7f1..a3072e3 100644 --- a/tests/integration/test_uow_integration.py +++ b/tests/integration/test_uow_integration.py @@ -3,6 +3,9 @@ from __future__ import annotations import logging +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from dataclasses import dataclass import pytest from sqlalchemy import func, select, text @@ -12,7 +15,7 @@ from sqlalchemy_foundation_kit.session.manager import AsyncSessionManager from sqlalchemy_foundation_kit.uow.enums import IsolationLevel from sqlalchemy_foundation_kit.uow.sqlalchemy import AsyncSQLAlchemyUnitOfWork, AsyncSQLAlchemyUowTransaction -from tests.integration.conftest import SEARCH_PATH_SCHEMA +from tests.integration.conftest import SEARCH_PATH_SCHEMA, PostgresContainer, asyncpg_url from tests.integration.models import TestUser @@ -422,3 +425,149 @@ async def test__transaction__failed_flush__row_values_reach_the_caller_and_no_lo assert email in str(raised.value) # ...and the library wrote none of it to the logs, at any level assert [r for r in caplog.records if r.name.startswith("sqlalchemy_foundation_kit")] == [] + + +# ============================================================================ +# Read-Only query() Tests (issue #56 scenario) +# ============================================================================ + +READ_ONLY_SQL_TRANSACTION = "25006" + + +@dataclass +class _OneConnection: + kind: str = "async_adapted_queue" + size: int = 1 + max_overflow: int = 0 + pre_ping: bool = False + recycle: int = -1 + timeout: float = 5.0 + + +def _sqlstate(error: DBAPIError) -> str | None: + return getattr(error.orig, "sqlstate", None) + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__query__insert_and_commit__rejected_and_nothing_persisted( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], + async_session: AsyncSession, +) -> None: + # Arrange: the reporter's write through query(), committed explicitly + email = "written-inside-query@example.com" + + # Act + with pytest.raises(DBAPIError) as raised: + async with uow.query() as qx: + qx.session.add(TestUser(name="Query", email=email, age=30)) + await qx.session.commit() + + # Assert + assert _sqlstate(raised.value) == READ_ONLY_SQL_TRANSACTION + result = await async_session.execute(select(TestUser).where(TestUser.email == email)) + assert result.scalar_one_or_none() is None + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__query__reads__run_in_a_read_only_transaction( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], +) -> None: + # Act + async with uow.query() as qx: + read_only = (await qx.session.execute(text("SHOW transaction_read_only"))).scalar_one() + users = (await qx.session.execute(select(func.count()).select_from(TestUser))).scalar_one() + + # Assert + assert read_only == "on" + assert users == 0 + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__query__isolation_level__read_only_as_well( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], +) -> None: + # Act: the isolation level checks the connection out first, so the block marks the + # transaction itself + async with uow.query(isolation_level=IsolationLevel.SERIALIZABLE) as qx: + isolation = await _transaction_isolation(qx.session) + read_only = (await qx.session.execute(text("SHOW transaction_read_only"))).scalar_one() + + # Assert + assert (isolation, read_only) == ("serializable", "on") + + +class _StatementFirstUnitOfWork(AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction]): + """The ``open_session`` override from its docstring: a statement before the block runs.""" + + @asynccontextmanager + async def open_session(self, isolation_level: IsolationLevel | str | None = None) -> AsyncIterator[AsyncSession]: + async with super().open_session(isolation_level) as session: + await session.execute(text("SELECT set_config('app.tenant_id', '7', true)")) + yield session + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__query__open_session_override_ran_a_statement__still_read_only( + async_session_factory: async_sessionmaker[AsyncSession], +) -> None: + # Arrange + uow = _StatementFirstUnitOfWork(async_session_factory, transaction_factory=AsyncSQLAlchemyUowTransaction) + + # Act + with pytest.raises(DBAPIError) as raised: + async with uow.query() as qx: + tenant = (await qx.session.execute(text("SELECT current_setting('app.tenant_id')"))).scalar_one() + assert tenant == "7" + await qx.session.execute(text("INSERT INTO test_users (name, email, age) VALUES ('x', 'x@x', 1)")) + + # Assert + assert _sqlstate(raised.value) == READ_ONLY_SQL_TRANSACTION + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__query__connection_back_in_the_pool__next_transaction_can_write( + postgres_container: PostgresContainer, +) -> None: + # Arrange: one connection, so the transaction after query() gets the very same one + manager: AsyncSessionManager[AsyncSession] = AsyncSessionManager( + asyncpg_url(postgres_container), + poolclass="async_adapted_queue", + pool_settings=_OneConnection(), + ) + uow = AsyncSQLAlchemyUnitOfWork(manager.session_maker, transaction_factory=AsyncSQLAlchemyUowTransaction) + + # Act + async with manager: + async with uow.query() as qx: + reader_pid = (await qx.session.execute(text("SELECT pg_backend_pid()"))).scalar_one() + async with uow.transaction() as tx: + writer_pid = (await tx.session.execute(text("SELECT pg_backend_pid()"))).scalar_one() + read_only = (await tx.session.execute(text("SHOW transaction_read_only"))).scalar_one() + tx.session.add(TestUser(name="After", email="after-query@example.com", age=40)) + + # Assert + assert writer_pid == reader_pid + assert read_only == "off" + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__query__advisory_lock__held_until_the_block_exits( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], + async_session_factory: async_sessionmaker[AsyncSession], +) -> None: + # Act + async with uow.query() as qx: + mine = (await qx.session.execute(text("SELECT pg_try_advisory_xact_lock(56)"))).scalar_one() + async with async_session_factory() as other: + theirs_meanwhile = (await other.execute(text("SELECT pg_try_advisory_xact_lock(56)"))).scalar_one() + async with async_session_factory() as other: + theirs_after = (await other.execute(text("SELECT pg_try_advisory_xact_lock(56)"))).scalar_one() + + # Assert + assert (mine, theirs_meanwhile, theirs_after) == (True, False, True) diff --git a/tests/unit/contrib/telemetry/test_uow.py b/tests/unit/contrib/telemetry/test_uow.py index 81c75c2..a818a51 100644 --- a/tests/unit/contrib/telemetry/test_uow.py +++ b/tests/unit/contrib/telemetry/test_uow.py @@ -330,6 +330,7 @@ async def test__traced_async_unit_of_work__query_with_isolation_level__sets_attr mock_transaction = MagicMock() mock_session = AsyncMock() mock_session.begin = MagicMock() + mock_session.in_transaction = MagicMock(return_value=True) mock_session.close = AsyncMock() mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock()