Draft a psycopg journal

Author: ryanhiebert

pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
 
 [project.optional-dependencies]
 django = ["django>=6.0", "django-cmd>=3.0"]
+psycopg = ["psycopg>=3.3"]
 
 [project.entry-points.djp]
 queueio = "queueio.django.djp"
@@ -50,6 +51,7 @@ dev = [
     "ruff>=0.9.6",
     "textual-dev>=1.7.0",
     "types-pika>=1.2.0b1",
+    "psycopg>=3.3.3",
 ]
 
 [tool.ruff.lint]

pyrightconfig.json

@@ -1,5 +1,6 @@
 {
   "pythonVersion": "3.14",
   "typeCheckingMode": "standard",
-  "exclude": ["site", ".venv"]
+  "exclude": ["site", ".venv"],
+  "ignore": ["queueio/psycopg"]
 }

queueio/freethreading_test.py

@@ -22,7 +22,11 @@ def test_all_queueio_modules_preserve_freethreading():
     for _, modname, _ in pkgutil.walk_packages(
         queueio.__path__, f"{queueio.__name__}."
     ):
-        if not modname.endswith("_test") and ".django" not in modname:
+        if (
+            not modname.endswith("_test")
+            and ".django" not in modname
+            and ".psycopg" not in modname
+        ):
             importlib.import_module(modname)
 
     assert sys._is_gil_enabled() is False

queueio/psycopg/__init__.py

@@ -0,0 +1,90 @@
+"""WARNING: This module is a work in progress.
+
+The PsycopgJournal implementation has not been extensively reviewed. It passes
+the base test suite but may contain bugs or design flaws. You shouldn't use this yet.
+"""
+
+from collections.abc import Iterator
+from contextlib import suppress
+from threading import Lock
+
+import psycopg
+
+from queueio.journal import Journal
+
+
+class PsycopgJournal(Journal):
+    @classmethod
+    def from_uri(cls, uri: str, /):
+        return cls(uri)
+
+    def __init__(self, uri: str):
+        self.__publish_conn = psycopg.connect(uri, autocommit=True)
+        self.__subscribe_conn = psycopg.connect(uri, autocommit=True)
+        self.__publish_lock = Lock()
+        self.__shutdown_lock = Lock()
+        self.__shutdown = False
+
+        # Ensure schema exists
+        self.__publish_conn.execute("""
+            CREATE TABLE IF NOT EXISTS queueio_journal (
+                id BIGSERIAL PRIMARY KEY,
+                body BYTEA NOT NULL,
+                created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+            )
+        """)
+
+        # LISTEN before SELECT to avoid race condition:
+        # any NOTIFY after LISTEN will be buffered, and the subsequent
+        # SELECT establishes our starting position.
+        self.__subscribe_conn.execute("LISTEN queueio_journal")
+        row = self.__subscribe_conn.execute(
+            "SELECT COALESCE(MAX(id), 0) FROM queueio_journal"
+        ).fetchone()
+        assert row is not None
+        self.__last_id: int = row[0]
+
+    def publish(self, message: bytes):
+        with self.__publish_lock, self.__publish_conn.transaction():
+            self.__publish_conn.execute(
+                t"INSERT INTO queueio_journal (body) VALUES ({message})"
+            )
+            self.__publish_conn.execute("NOTIFY queueio_journal")
+
+    def subscribe(self) -> Iterator[bytes]:
+        try:
+            while not self.__shutdown:
+                rows = self.__subscribe_conn.execute(
+                    t"""
+                    SELECT id, body FROM queueio_journal
+                    WHERE id > {self.__last_id}
+                    ORDER BY id
+                    """
+                ).fetchall()
+
+                for row in rows:
+                    self.__last_id = row[0]
+                    yield row[1]
+
+                if not rows and not self.__shutdown:
+                    # Wait for a notification to wake us up
+                    for _ in self.__subscribe_conn.notifies(timeout=1.0):
+                        break  # Got one, go back to query loop
+        finally:
+            with suppress(Exception):
+                self.__subscribe_conn.execute("UNLISTEN queueio_journal")
+            with suppress(Exception):
+                self.__subscribe_conn.close()
+
+    def shutdown(self):
+        with self.__shutdown_lock:
+            if self.__shutdown:
+                return
+            self.__shutdown = True
+
+            # Wake up the subscriber by sending a notification
+            with suppress(Exception):
+                self.__publish_conn.execute("NOTIFY queueio_journal")
+
+            with suppress(Exception):
+                self.__publish_conn.close()

queueio/psycopg/journal.py

@@ -0,0 +1,42 @@
+import os
+
+import pytest
+
+psycopg = pytest.importorskip(
+    "psycopg", reason="psycopg not available", exc_type=ImportError
+)
+
+from queueio.journal_test import BaseJournalTest  # noqa: E402
+
+from .journal import PsycopgJournal  # noqa: E402
+
+PSYCOPG_TEST_URI = os.environ.get(
+    "QUEUEIO_PSYCOPG_TEST", "postgresql://postgres@localhost/queueio_test"
+)
+
+
+def psycopg_available() -> bool:
+    try:
+        conn = psycopg.connect(PSYCOPG_TEST_URI)
+        conn.close()
+        return True
+    except Exception:
+        return False
+
+
+pytestmark = pytest.mark.skipif(
+    not psycopg_available(), reason="PostgreSQL not available"
+)
+
+
+class TestPsycopgJournal(BaseJournalTest):
+    @pytest.fixture
+    def journal(self):
+        journal = PsycopgJournal.from_uri(PSYCOPG_TEST_URI)
+
+        # Truncate to isolate tests
+        with psycopg.connect(PSYCOPG_TEST_URI, autocommit=True) as conn:
+            conn.execute("TRUNCATE queueio_journal")
+
+        yield journal
+        journal.shutdown()

queueio/psycopg/journal_test.py

@@ -140,7 +140,9 @@ def __default_journal(self) -> Journal:
                 )
             if not (uri.startswith("postgresql:") or uri.startswith("postgres:")):
                 raise ValueError(f"URI scheme must be 'postgresql:', got: {uri}")
-            raise ValueError("Journal backend 'psycopg' is not yet implemented.")
+            from .psycopg.journal import PsycopgJournal
+
+            return PsycopgJournal.from_uri(uri)
 
         raise ValueError(f"Unknown journal backend: {backend}")