feat: accept an AsyncConnection wherever publishing takes a session

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():

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.
 

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,

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)

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]:

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

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.
 """
 

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,

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,