From 7cc9897ab2485f032317595177a3bc58358ecd25 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 21:37:18 +0300 Subject: [PATCH] feat: accept an AsyncConnection wherever publishing takes a session broker.publish, publish_batch, cancel_timer, fetch_unprocessed, OutboxPublisher.publish and OutboxResponse now take a SQLAlchemy Core AsyncConnection through the existing required session= keyword, so code without the ORM can publish on the connection its domain writes run on. NOTIFY dedup keys on the connection's own transaction, so savepoints and autobegin behave as they do on a session. --- docs/introduction/how-it-works.md | 4 +- docs/usage/basic.md | 4 +- docs/usage/publisher.md | 29 ++++-- faststream_outbox/_session.py | 12 +++ faststream_outbox/broker.py | 15 ++-- faststream_outbox/client.py | 50 +++++------ faststream_outbox/publisher/producer.py | 2 +- faststream_outbox/publisher/usecase.py | 4 +- faststream_outbox/response.py | 14 ++- faststream_outbox/testing.py | 9 +- tests/test_fake.py | 16 +++- tests/test_integration.py | 113 ++++++++++++++++++++++++ tests/test_unit.py | 12 +-- 13 files changed, 220 insertions(+), 64 deletions(-) create mode 100644 faststream_outbox/_session.py diff --git a/docs/introduction/how-it-works.md b/docs/introduction/how-it-works.md index bcaceab..4038d01 100644 --- a/docs/introduction/how-it-works.md +++ b/docs/introduction/how-it-works.md @@ -34,8 +34,8 @@ transactions are the better fit.* ## Producer side `broker.publish(body, *, queue, session, ...)` inserts an outbox row through -the caller's `AsyncSession`. It does not flush, commit, or open its own -transaction; the row must commit with the caller's domain writes: +the caller's `AsyncSession` (or `AsyncConnection`, for Core code). It does not +flush, commit, or open its own transaction; the row must commit with the caller's domain writes: ```python async with session_factory() as session, session.begin(): diff --git a/docs/usage/basic.md b/docs/usage/basic.md index b553647..9fb8beb 100644 --- a/docs/usage/basic.md +++ b/docs/usage/basic.md @@ -64,7 +64,9 @@ async with session_factory() as session, session.begin(): # session.begin() commits both atomically on exit ``` -Passing anything that is not an `AsyncSession` raises `TypeError`. The whole +Code that uses SQLAlchemy Core without the ORM can pass its `AsyncConnection` +instead; see [Publishing on a Core connection](./publisher.md#publishing-on-a-core-connection). +Passing anything else raises `TypeError`. The whole point of the outbox pattern is that the row commits atomically with your domain writes; opening a separate session would defeat it. diff --git a/docs/usage/publisher.md b/docs/usage/publisher.md index 7313c73..e7d6c6b 100644 --- a/docs/usage/publisher.md +++ b/docs/usage/publisher.md @@ -8,8 +8,8 @@ There are three ways to write an outbox row: per-queue config and AsyncAPI spec coverage. All three share the same transactional contract: the caller supplies an -`AsyncSession`, and the row commits with the caller's domain writes. The -broker does not flush, commit, or open its own transaction. +`AsyncSession` or an `AsyncConnection`, and the row commits with the caller's +domain writes. The broker does not flush, commit, or open its own transaction. For "consume from A → enqueue to B" relay flows, a fourth path is available: returning `OutboxResponse(...)` from a handler. See [Chained @@ -37,7 +37,7 @@ await broker.publish( body, *, queue: str, - session: AsyncSession, + session: AsyncSession | AsyncConnection, headers: dict[str, str] | None = None, correlation_id: str | None = None, activate_in: timedelta | None = None, @@ -50,7 +50,26 @@ Returns the inserted row's `id`, or `None` if a `timer_id` conflict made the insert a no-op. See [Timers](./timers.md) for `activate_in` / `activate_at` / `timer_id`. -Passing anything that is not an `AsyncSession` raises `TypeError`. +Passing anything other than an `AsyncSession` or `AsyncConnection` raises +`TypeError`. + +### Publishing on a Core connection + +If your code uses SQLAlchemy Core without the ORM, pass the `AsyncConnection` +your domain writes run on. The parameter is still called `session`: + +```python +async with engine.begin() as conn: + await conn.execute(insert(orders).values(id=1)) # domain write + await broker.publish({"order_id": 1}, queue="orders", session=conn) +``` + +Everything else works the same: the row commits or rolls back with `conn`'s +transaction, a savepoint from `conn.begin_nested()` scopes it like a session +savepoint, and a connection opened with `engine.connect()` without `begin()` +starts its transaction on the first publish, so you still have to call +`await conn.commit()`. `publish_batch`, `cancel_timer`, `fetch_unprocessed`, +`OutboxPublisher.publish` and `OutboxResponse` accept a connection too. ## `broker.publish_batch` @@ -73,7 +92,7 @@ Full signature: await broker.publish_batch( *bodies, queue: str, - session: AsyncSession, + session: AsyncSession | AsyncConnection, headers: dict[str, str] | None = None, activate_in: timedelta | None = None, activate_at: datetime | None = None, diff --git a/faststream_outbox/_session.py b/faststream_outbox/_session.py new file mode 100644 index 0000000..1faf694 --- /dev/null +++ b/faststream_outbox/_session.py @@ -0,0 +1,12 @@ +"""The caller-owned handle an outbox row is written on; a runtime union, so it also works with ``isinstance``.""" + +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession + + +SessionOrConnection = AsyncSession | AsyncConnection + + +def require_session_or_connection(session: object, context: str) -> None: + if not isinstance(session, SessionOrConnection): + msg = f"{context} requires an sqlalchemy.ext.asyncio.AsyncSession or AsyncConnection" + raise TypeError(msg) diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index f460c8f..2a7d748 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -26,10 +26,10 @@ from faststream.exceptions import IncorrectState from faststream.specification.schema import BrokerSpec from faststream.specification.schema.extra import Tag, TagDict -from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.pool import QueuePool from typing_extensions import override +from faststream_outbox._session import SessionOrConnection, require_session_or_connection from faststream_outbox.client import AbstractOutboxClient, OutboxClient from faststream_outbox.configs import LastExceptionRenderer, OutboxBrokerConfig from faststream_outbox.message import OutboxInnerMessage @@ -413,7 +413,7 @@ async def publish( # ty: ignore[invalid-method-override] body: typing.Any, *, queue: str, - session: AsyncSession, + session: SessionOrConnection, headers: dict[str, str] | None = None, correlation_id: str | None = None, activate_in: dt.timedelta | None = None, @@ -422,6 +422,7 @@ async def publish( # ty: ignore[invalid-method-override] ) -> int | None: """Insert one outbox row using *session*'s open transaction. + *session* is an ``AsyncSession`` or, for Core code, an ``AsyncConnection``. Must be called inside a transaction the caller owns (typically inside an ``async with session.begin():`` block). ``publish`` does not flush, commit, or open its own transaction — that is the whole point of the transactional @@ -453,7 +454,7 @@ async def publish_batch( # ty: ignore[invalid-method-override] self, *bodies: typing.Any, queue: str, - session: AsyncSession, + session: SessionOrConnection, headers: dict[str, str] | None = None, activate_in: dt.timedelta | None = None, activate_at: dt.datetime | None = None, @@ -467,9 +468,7 @@ async def publish_batch( # ty: ignore[invalid-method-override] """ # P1: validate the session type up front so an empty batch fails the same way a # non-empty one does (the command constructor that checks it is only built below). - if not isinstance(session, AsyncSession): - msg = "broker.publish_batch requires an sqlalchemy.ext.asyncio.AsyncSession" - raise TypeError(msg) + require_session_or_connection(session, "broker.publish_batch") if not bodies: # Validate queue + activate args even when there's no work so an empty # batch rejects the same misconfigurations a non-empty one does (F4-06). @@ -499,7 +498,7 @@ async def cancel_timer( *, queue: str, timer_id: str, - session: AsyncSession, + session: SessionOrConnection, ) -> bool: """Delete a not-yet-leased timer row. Idempotent at the ``(queue, timer_id)`` level. @@ -534,7 +533,7 @@ async def cancel_timer( async def fetch_unprocessed( self, *, - session: AsyncSession, + session: SessionOrConnection, queue: str | None = None, limit: int = 1000, ) -> list[OutboxInnerMessage]: diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index 636b828..6b87420 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -44,6 +44,7 @@ from faststream_outbox import schema_validation from faststream_outbox._scheduling import is_future_dated, resolve_next_attempt_client_side +from faststream_outbox._session import SessionOrConnection, require_session_or_connection from faststream_outbox._time import utcnow from faststream_outbox.message import OutboxInnerMessage from faststream_outbox.schema import ( @@ -90,7 +91,7 @@ def table(self) -> "Table": ... @abc.abstractmethod async def insert( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, payload: bytes, @@ -109,7 +110,7 @@ async def insert( @abc.abstractmethod async def insert_batch( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, rows: "Sequence[tuple[bytes, dict[str, str] | None]]", @@ -163,13 +164,13 @@ async def mark_pending_with_lease( ) -> bool: ... @abc.abstractmethod - async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSession") -> bool: ... + async def cancel_timer(self, *, queue: str, timer_id: str, session: SessionOrConnection) -> bool: ... @abc.abstractmethod async def fetch_unprocessed( self, *, - session: "AsyncSession", + session: SessionOrConnection, queue: str | None = None, limit: int = 1000, ) -> list[OutboxInnerMessage]: ... @@ -282,7 +283,7 @@ def __init__(self, table: "Table") -> None: async def insert( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, payload: bytes, @@ -316,7 +317,7 @@ async def insert( async def insert_batch( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, rows: "Sequence[tuple[bytes, dict[str, str] | None]]", @@ -339,15 +340,15 @@ async def insert_batch( if not is_future_dated(activate_in, activate_at, now): await self._notify(session, queue) - def _notified_queues(self, session: "AsyncSession") -> set[str] | None: + def _notified_queues(self, session: SessionOrConnection) -> set[str] | None: """Return the queues already notified in *session*'s innermost transaction, or None outside one.""" - sync_session = session.sync_session - txn = sync_session.get_nested_transaction() or sync_session.get_transaction() + sync_owner = session.sync_session if isinstance(session, AsyncSession) else session.sync_connection + txn = None if sync_owner is None else sync_owner.get_nested_transaction() or sync_owner.get_transaction() if txn is None: return None return self._notified.setdefault(txn, set()) - async def _notify(self, session: "AsyncSession", queue: str) -> None: + async def _notify(self, session: SessionOrConnection, queue: str) -> None: # ``pg_notify(:channel, :payload)`` -- parameterized so channel and payload # bind cleanly (raw NOTIFY accepts only literals -- injection-prone). Runs on # the caller's session so NOTIFY commits with the row insert; rollback silently @@ -359,12 +360,13 @@ async def _notify(self, session: "AsyncSession", queue: str) -> None: # active transaction (get_nested_transaction() during a savepoint, else the outer # transaction) -- each a distinct, weak-referenceable object that GCs when the # transaction ends, so a rolled-back savepoint cannot suppress a later real NOTIFY. - # Key on the *sync* SessionTransaction, not the async AsyncSessionTransaction proxy: - # SQLAlchemy regenerates that proxy on every call and only weak-references it, so - # under autobegin (no explicit `session.begin()`) it would GC between publishes and - # defeat the memo. The Session holds the sync transaction strongly for its lifetime, - # so the key survives across publishes whether the caller uses `session.begin()` or - # autobegin. No active transaction -> emit unconditionally. + # Key on the *sync* transaction (the Session's SessionTransaction, or the Connection's + # RootTransaction / NestedTransaction), not the async proxy around it: SQLAlchemy + # regenerates that proxy on every call and only weak-references it, so under autobegin + # (no explicit `begin()`) it would GC between publishes and defeat the memo. The sync + # Session or Connection holds its transaction strongly while it is open, so the key + # survives across publishes whether the caller uses `begin()` or autobegin. No active + # transaction -> emit unconditionally. emitted = self._notified_queues(session) if emitted is not None: if queue in emitted: @@ -410,7 +412,7 @@ def engine(self) -> "AsyncEngine": async def insert( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, payload: bytes, @@ -431,7 +433,7 @@ async def insert( async def insert_batch( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, rows: "Sequence[tuple[bytes, dict[str, str] | None]]", @@ -618,11 +620,9 @@ async def mark_pending_with_lease( result = await conn.execute(stmt, {"delay": max(0.0, delay_seconds)}) return (result.rowcount or 0) > 0 - async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSession") -> bool: + async def cancel_timer(self, *, queue: str, timer_id: str, session: SessionOrConnection) -> bool: """Delete a not-yet-leased ``(queue, timer_id)`` row on the caller's session.""" - if not isinstance(session, AsyncSession): - msg = "OutboxClient.cancel_timer requires an sqlalchemy.ext.asyncio.AsyncSession" - raise TypeError(msg) + require_session_or_connection(session, "OutboxClient.cancel_timer") t = self._table stmt = delete(t).where( t.c.queue == queue, @@ -635,14 +635,12 @@ async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSessio async def fetch_unprocessed( self, *, - session: "AsyncSession", + session: SessionOrConnection, queue: str | None = None, limit: int = 1000, ) -> list[OutboxInnerMessage]: """Read up to *limit* rows (optionally filtered by *queue*) on the caller's session.""" - if not isinstance(session, AsyncSession): - msg = "OutboxClient.fetch_unprocessed requires an sqlalchemy.ext.asyncio.AsyncSession" - raise TypeError(msg) + require_session_or_connection(session, "OutboxClient.fetch_unprocessed") if limit < 1: # F4-04: a non-positive limit otherwise hits SQL (LIMIT -1 → DB error) or # silently returns nothing (LIMIT 0); reject it up front, consistently with diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py index 382f02c..6bd31eb 100644 --- a/faststream_outbox/publisher/producer.py +++ b/faststream_outbox/publisher/producer.py @@ -4,7 +4,7 @@ route through this producer via FastStream's ``_basic_publish(cmd, producer=...)`` middleware stack, so encode → insert → ``published`` semantics live in one place. The insert itself (and its NOTIFY) belongs to the broker's client. The producer never opens -its own session: every command carries the caller's ``AsyncSession`` so the row commits +its own session: every command carries the caller's ``AsyncSession`` or ``AsyncConnection`` so the row commits atomically with the caller's domain writes. """ diff --git a/faststream_outbox/publisher/usecase.py b/faststream_outbox/publisher/usecase.py index 9e2932c..4e52ba2 100644 --- a/faststream_outbox/publisher/usecase.py +++ b/faststream_outbox/publisher/usecase.py @@ -12,9 +12,9 @@ import typing from faststream._internal.endpoint.publisher import PublisherUsecase -from sqlalchemy.ext.asyncio import AsyncSession from typing_extensions import override +from faststream_outbox._session import SessionOrConnection from faststream_outbox.publisher.config import OutboxPublisherConfig from faststream_outbox.publisher.specification import OutboxPublisherSpecification from faststream_outbox.response import _REQUEST_UNSUPPORTED_MSG, OutboxPublishCommand @@ -68,7 +68,7 @@ async def publish( # ty: ignore[invalid-method-override] self, body: typing.Any, # parity with broker.publish (both build OutboxPublishCommand(body: Any)) *, - session: AsyncSession, + session: SessionOrConnection, headers: dict[str, str] | None = None, correlation_id: str | None = None, activate_in: dt.timedelta | None = None, diff --git a/faststream_outbox/response.py b/faststream_outbox/response.py index 1af3844..dcf6ed5 100644 --- a/faststream_outbox/response.py +++ b/faststream_outbox/response.py @@ -1,8 +1,8 @@ """Publish-side DTO for the outbox transport. ``OutboxPublishCommand`` carries the domain fields that travel from the -publisher / broker into ``OutboxProducer.publish``: the user's ``AsyncSession`` -(required — outbox rows commit with the caller's transaction), the queue name, +publisher / broker into ``OutboxProducer.publish``: the user's ``AsyncSession`` or +``AsyncConnection`` (required — outbox rows commit with the caller's transaction), the queue name, scheduling args, the timer dedup key. Centralizing the activate-args validation in the constructor ensures ``broker.publish``, ``broker.publish_batch`` and ``OutboxPublisher.publish`` all reject the same misconfigurations. @@ -17,8 +17,8 @@ from faststream.response.publish_type import PublishType from faststream.response.response import BatchPublishCommand, PublishCommand, Response -from sqlalchemy.ext.asyncio import AsyncSession +from faststream_outbox._session import SessionOrConnection, require_session_or_connection from faststream_outbox.message import CORRELATION_ID_HEADER @@ -55,9 +55,7 @@ def _validate_publish_args( if activate_at is not None and activate_at.tzinfo is None: msg = f"{context} requires activate_at to be timezone-aware" raise ValueError(msg) - if not isinstance(session, AsyncSession): - msg = f"{context} requires an sqlalchemy.ext.asyncio.AsyncSession" - raise TypeError(msg) + require_session_or_connection(session, context) # ``queue`` reaches SQL otherwise unvalidated — empty / non-str / over the # ``String(255)`` column would surface as an opaque DB error or silent truncation. if not isinstance(queue, str): @@ -91,7 +89,7 @@ def __init__( /, *bodies: typing.Any, queue: str, - session: AsyncSession, + session: SessionOrConnection, headers: dict[str, str] | None = None, correlation_id: str | None = None, activate_in: dt.timedelta | None = None, @@ -195,7 +193,7 @@ def __init__( body: typing.Any, *, queue: str, - session: AsyncSession, + session: SessionOrConnection, headers: dict[str, str] | None = None, correlation_id: str | None = None, activate_in: dt.timedelta | None = None, diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py index 339f392..fe4d3c8 100644 --- a/faststream_outbox/testing.py +++ b/faststream_outbox/testing.py @@ -25,6 +25,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from faststream_outbox._scheduling import resolve_next_attempt_client_side +from faststream_outbox._session import SessionOrConnection from faststream_outbox._time import utcnow from faststream_outbox.broker import OutboxBroker from faststream_outbox.client import _NO_DLQ_TABLE_MSG, AbstractOutboxClient @@ -139,7 +140,7 @@ def feed( async def insert( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, payload: bytes, @@ -164,7 +165,7 @@ async def insert( async def insert_batch( self, - session: "AsyncSession", + session: SessionOrConnection, *, queue: str, rows: "Sequence[tuple[bytes, dict[str, str] | None]]", @@ -341,7 +342,7 @@ async def mark_pending_with_lease( return True return False - async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSession") -> bool: + async def cancel_timer(self, *, queue: str, timer_id: str, session: SessionOrConnection) -> bool: """Mirror :meth:`OutboxBroker.cancel_timer` — drop a not-yet-leased timer row.""" del session # ignored (no real DB), matching the fake publish contract for i, row in enumerate(self._rows): @@ -353,7 +354,7 @@ async def cancel_timer(self, *, queue: str, timer_id: str, session: "AsyncSessio async def fetch_unprocessed( self, *, - session: "AsyncSession", + session: SessionOrConnection, queue: str | None = None, limit: int = 1000, ) -> list[OutboxInnerMessage]: diff --git a/tests/test_fake.py b/tests/test_fake.py index d7abde2..df72ba6 100644 --- a/tests/test_fake.py +++ b/tests/test_fake.py @@ -14,7 +14,7 @@ from faststream.exceptions import NackMessage, StopConsume from faststream.middlewares import AckPolicy from sqlalchemy import MetaData -from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession from typing_extensions import get_protocol_members from faststream_outbox import ( @@ -110,6 +110,20 @@ async def handle(body: dict) -> None: assert test_broker.fake_client.rows == [] # row deleted after ack +async def test_publish_on_a_core_connection_dispatches_under_the_test_broker() -> None: + broker = _make_broker() + received: list[dict] = [] + + @broker.subscriber("orders") + async def handle(body: dict) -> None: + received.append(body) + + async with TestOutboxBroker(broker): + await broker.publish({"order_id": 1}, queue="orders", session=AsyncMock(spec=AsyncConnection)) + + assert received == [{"order_id": 1}] + + async def test_sync_batch_inserts_whole_batch_before_dispatch_and_published_first() -> None: """S5: sync-mode batch inserts the whole batch before any handler runs, and ``published`` precedes delivery.""" events: list[str] = [] diff --git a/tests/test_integration.py b/tests/test_integration.py index e69be4e..1240d4e 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2706,3 +2706,116 @@ async def test_notify_deduped_in_autobegin_transaction(pg_engine, outbox_table) await broker.publish({"n": 1}, queue="orders", session=session) await session.commit() assert count[0] == 1 # 5 publishes, one queue, one autobegin transaction -> one pg_notify + + +async def test_publish_on_a_core_connection_commits_with_its_transaction( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + async with pg_engine.begin() as conn: + rid = await broker.publish({"n": 1}, queue="orders", session=conn) + assert await _row_count(pg_engine, outbox_table) == 0 + assert rid is not None + assert await _row_count(pg_engine, outbox_table) == 1 + + +async def test_publish_on_a_core_connection_rolls_back_with_its_transaction( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + async with pg_engine.connect() as conn: + trans = await conn.begin() + await broker.publish({"n": 1}, queue="orders", session=conn) + await trans.rollback() + assert await _row_count(pg_engine, outbox_table) == 0 + + +async def test_publish_on_an_autobegun_core_connection_lands_on_commit( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + with _count_pg_notify(pg_engine) as count: + async with pg_engine.connect() as conn: + for _ in range(5): + await broker.publish({"n": 1}, queue="orders", session=conn) + assert await _row_count(pg_engine, outbox_table) == 0 + await conn.commit() + assert await _row_count(pg_engine, outbox_table) == 5 + assert count[0] == 1 + + +async def test_notify_on_a_core_connection_re_emits_after_savepoint_rollback( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + + class _RollbackError(Exception): + pass + + with _count_pg_notify(pg_engine) as count: + async with pg_engine.begin() as conn: + with contextlib.suppress(_RollbackError): + async with conn.begin_nested(): + await broker.publish({"n": 1}, queue="orders", session=conn) + raise _RollbackError + count[0] = 0 + await broker.publish({"n": 2}, queue="orders", session=conn) + await broker.publish({"n": 3}, queue="orders", session=conn) + assert count[0] == 1 + + +async def test_publish_batch_on_a_core_connection_shares_the_notify_dedup( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + with _count_pg_notify(pg_engine) as count: + async with pg_engine.begin() as conn: + await broker.publish_batch({"n": 1}, {"n": 2}, queue="orders", session=conn) + await broker.publish({"n": 3}, queue="orders", session=conn) + assert await _row_count(pg_engine, outbox_table) == 3 + assert count[0] == 1 + + +async def test_cancel_timer_on_a_core_connection_deletes_the_pending_timer( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + async with pg_engine.begin() as conn: + await broker.publish( + {"n": 1}, queue="orders", session=conn, activate_in=dt.timedelta(minutes=5), timer_id="t-1" + ) + async with pg_engine.begin() as conn: + assert await broker.cancel_timer(queue="orders", timer_id="t-1", session=conn) is True + assert await _row_count(pg_engine, outbox_table) == 0 + + +async def test_fetch_unprocessed_on_a_core_connection_sees_its_uncommitted_rows( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + async with pg_engine.begin() as conn: + await broker.publish("pre-commit", queue="orders", session=conn) + rows = await broker.fetch_unprocessed(session=conn, queue="orders") + assert [r.queue for r in rows] == ["orders"] + + +async def test_outbox_response_on_a_core_connection_commits_with_its_transaction( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + response_publisher = OutboxFakePublisher(broker.config.producer) + async with pg_engine.begin() as conn: + response = OutboxResponse(body={"chained": True}, queue="downstream", session=conn) + await response_publisher._publish(response.as_publish_command(), _extra_middlewares=[]) # noqa: SLF001 + assert await _row_count(pg_engine, outbox_table) == 0 + assert await _row_count(pg_engine, outbox_table) == 1 + + +async def test_publisher_publish_on_a_core_connection_commits_with_its_transaction( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + pub = broker.publisher("orders") + async with pg_engine.begin() as conn: + assert await pub.publish({"n": 1}, session=conn) is not None + assert await _row_count(pg_engine, outbox_table) == 1 diff --git a/tests/test_unit.py b/tests/test_unit.py index 5f8ff8d..7f0fb25 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -598,15 +598,15 @@ def test_broker_with_engine_has_client() -> None: async def test_broker_publish_rejects_non_async_session() -> None: - """INVARIANT: a non-``AsyncSession`` session argument is refused at the publish boundary. + """INVARIANT: a session that is neither ``AsyncSession`` nor ``AsyncConnection`` is refused at the boundary. - The transactional contract is only enforceable through a real session; anything else silently - degrades it. Accepting a duck-typed object (a sync ``Session``, a connection, a mock in - production code) would let the insert run outside the caller's transaction and fail as lost - messages rather than as a ``TypeError`` at the call site. + The transactional contract is only enforceable through a handle that owns the caller's transaction; + anything else silently degrades it. Accepting a duck-typed object (a sync ``Session``, a raw driver + connection, a mock in production code) would let the insert run outside the caller's transaction and + fail as lost messages rather than as a ``TypeError`` at the call site. """ broker = _make_broker() - with pytest.raises(TypeError, match="AsyncSession"): + with pytest.raises(TypeError, match="AsyncSession or AsyncConnection"): await broker.publish(b"x", queue="orders", session=object()) # ty: ignore[invalid-argument-type]