How to enqueue a pgqueuer job in the same SQLAlchemy transaction?

[yqiang]

I have a FastAPI app using SQLAlchemy, and each request gets a request-scoped SQLAlchemy Session dependency.

I want to enqueue a pgqueuer job in the same database transaction as my application writes, so that if the SQLAlchemy session rolls back, the pgqueuer job is also rolled back.

From the docs, the examples I found create a Queries object from an asyncpg connection or pool, which looks like a separate connection/transaction from the SQLAlchemy session.

Is there a way to enqueue a job using an existing SQLAlchemy Session / transaction?

My goal is transactional enqueue semantics, not just “write app data and then enqueue separately”.

[janbjorge]

Hey, this is on our radar but we're in the middle of a large refactor heading toward v1 so it won't land in the library anytime soon.

In the meantime, here's a self-contained shim you can drop into your own project. Do not wire this up as a full PgQueuer driver — it has no LISTEN/NOTIFY support and will blow up if anything tries to use it for consuming.
It is strictly for enqueuing inside an existing transaction.

from __future__ import annotations

import asyncio
from typing import Any, Callable

import psycopg
from psycopg import AsyncRawCursor
from psycopg.rows import dict_row
from typing_extensions import Self

from pgqueuer.adapters.persistence.queries import Queries
from pgqueuer.core.tm import TaskManager


class PsycopgTransactionalDriver:
    """Enqueue-only driver for a psycopg connection already inside a transaction.

    Do NOT use this as a general PgQueuer driver. It has no LISTEN/NOTIFY
    support and cannot be used for consuming jobs.
    """

    def __init__(self, connection: psycopg.AsyncConnection) -> None:
        self._connection = connection
        self._shutdown = asyncio.Event()
        self._tm = TaskManager()

    @property
    def shutdown(self) -> asyncio.Event:
        return self._shutdown

    @property
    def tm(self) -> TaskManager:
        return self._tm

    async def fetch(self, query: str, *args: Any) -> list[dict]:
        cursor = AsyncRawCursor(self._connection, row_factory=dict_row)
        await cursor.execute(query, args or None)
        return await cursor.fetchall()

    async def execute(self, query: str, *args: Any) -> str:
        cursor = AsyncRawCursor(self._connection)
        await cursor.execute(query, args or None)
        return cursor.statusmessage or ""

    async def add_listener(
        self,
        channel: str,
        callback: Callable[[str | bytes | bytearray], None],
    ) -> None:
        raise RuntimeError("enqueue-only driver — consumer operations not supported")

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(self, *_: object) -> None:
        pass

Usage with SQLAlchemy:

async with AsyncSession(engine) as session:
    async with session.begin():
        bind = await session.connection()
        raw  = await bind.get_raw_connection()
        q = Queries(PsycopgTransactionalDriver(raw.driver_connection))
        await q.enqueue("send_email", payload=b"...")
        session.add(MyModel(...))
    # both commit or both roll back together

Note: this assumes your engine uses the postgresql+psycopg dialect. If you're on postgresql+asyncpg the object behind raw.driver_connection will be an asyncpg.Connection and this shim won't work as-is.

The reason this works: enqueue is a plain INSERT CTE, and Postgres defers the pg_notify() fired by the queue trigger until commit. So if the transaction rolls back, the notification is also discarded and no worker is ever
woken up.

[yqiang]

Thanks -- as an alternative, I found the PgRest integration documentation which defines a fn_pgqueuer_enqueue postgres function that inserts the job. Any downsides to just using that function (or the equivalent of it) directly?

[janbjorge]

I think as of now, there is no difference. But if there are changes to the enqueue in the future the PsycopgTransactionalDriver based approach is probably more future proof.