query() is documented as read-only and transaction-free, but it holds a transaction for the whole block and lets writes commit #56

import asyncio

from sqlalchemy import text
from sqlalchemy.exc import DBAPIError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy_foundation_kit import AsyncSQLAlchemyUnitOfWork, AsyncSQLAlchemyUowTransaction
from testcontainers.postgres import PostgresContainer


async def main(url: str) -> None:
    engine = create_async_engine(url)
    sessions = async_sessionmaker(engine, expire_on_commit=False)
    uow = AsyncSQLAlchemyUnitOfWork(sessions, AsyncSQLAlchemyUowTransaction)
    async with engine.begin() as conn:
        await conn.execute(text("CREATE TABLE notes (body text)"))

    # 1. "does not start a transaction"
    async with uow.query() as qx:
        pid = await qx.session.scalar(text("SELECT pg_backend_pid()"))
        await qx.session.execute(text("SELECT count(*) FROM notes"))
        same_now = await qx.session.scalar(text("SELECT now() = statement_timestamp()"))
        async with sessions() as observer:
            state = await observer.scalar(
                text("SELECT state FROM pg_stat_activity WHERE pid = :pid"), {"pid": pid}
            )
            locks = (await observer.execute(text(
                "SELECT mode FROM pg_locks WHERE pid = :pid AND relation = 'notes'::regclass"
            ), {"pid": pid})).scalars().all()
    print(f"1. inside query(): server state = {state!r}, now() = statement_timestamp(): {same_now}, "
          f"locks still held on notes: {locks}")

    # 2. a transaction-scoped advisory lock taken in query() is held until the block exits
    async with uow.query() as qx:
        mine = await qx.session.scalar(text("SELECT pg_try_advisory_xact_lock(42)"))
        async with sessions() as other:
            theirs = await other.scalar(text("SELECT pg_try_advisory_xact_lock(42)"))
    print(f"2. advisory lock in query(): acquired={mine}; another session's attempt meanwhile={theirs}")

    # 3. "read-only": an INSERT plus an explicit commit is persisted
    async with uow.query() as qx:
        await qx.session.execute(text("INSERT INTO notes VALUES ('written inside query()')"))
        await qx.session.commit()
    async with sessions() as s:
        rows = await s.scalar(text("SELECT count(*) FROM notes"))
    print(f"3. rows after INSERT + commit inside query(): {rows}")
    await engine.dispose()

    # 4. the proposal: begin the block's transaction READ ONLY (one-connection pool to see the reset)
    single_engine = create_async_engine(url, pool_size=1, max_overflow=0)
    single = async_sessionmaker(single_engine)
    async with single() as s:
        await s.connection(execution_options={"postgresql_readonly": True})
        pid = await s.scalar(text("SELECT pg_backend_pid()"))
        read_only = await s.scalar(text("SHOW transaction_read_only"))
        try:
            await s.execute(text("INSERT INTO notes VALUES ('x')"))
            outcome = "accepted"
        except DBAPIError as exc:
            outcome = f"rejected, SQLSTATE {exc.orig.sqlstate}"
    async with single() as s:
        next_pid = await s.scalar(text("SELECT pg_backend_pid()"))
        next_read_only = await s.scalar(text("SHOW transaction_read_only"))
    print(f"4. postgresql_readonly: transaction_read_only={read_only}, INSERT {outcome}; "
          f"next session, same connection={pid == next_pid}: transaction_read_only={next_read_only}")

    # 5. the same after the transaction has already run a statement (e.g. from an open_session() override)
    async with single() as s:
        await s.execute(text("SELECT 1"))
        await s.execute(text("SET TRANSACTION READ ONLY"))
        try:
            await s.execute(text("INSERT INTO notes VALUES ('x')"))
            outcome = "accepted"
        except DBAPIError as exc:
            outcome = f"rejected, SQLSTATE {exc.orig.sqlstate}"
    print(f"5. SET TRANSACTION READ ONLY after a first statement: INSERT {outcome}")
    await single_engine.dispose()


with PostgresContainer("postgres:17-alpine", driver="asyncpg") as pg:
    asyncio.run(main(pg.get_connection_url()))