fix(python): make job schedulers advance and support the postgres backend

.github/workflows/test.yml

@@ -357,10 +357,13 @@ jobs:
 
   postgres:
     runs-on: ubuntu-latest
-    needs: ioredis-redis
+    needs: [changes, ioredis-redis]
 
     name: testing node@${{ matrix.node-version }}, postgres@${{ matrix.postgres-version }}
 
+    env:
+      should_run: ${{ needs.changes.outputs.node == 'true' || needs.changes.outputs.lua == 'true' }}
+
     strategy:
       matrix:
         node-version: [lts/*]
@@ -386,18 +389,26 @@ jobs:
           --health-retries 5
 
     steps:
+      - name: Skip when node/lua inputs are unchanged
+        if: env.should_run != 'true'
+        run: echo "No node/lua changes detected; skipping postgres test steps."
       - name: Checkout repository
+        if: env.should_run == 'true'
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       - name: Use Node.js ${{ matrix.node-version }}
+        if: env.should_run == 'true'
         uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
         with:
           node-version: ${{ matrix.node-version }}
           cache: 'yarn'
-      - run: yarn install --ignore-engines --frozen-lockfile --non-interactive
+      - if: env.should_run == 'true'
+        run: yarn install --ignore-engines --frozen-lockfile --non-interactive
       # `yarn build` runs the pretest codegen (raw scripts, command transforms)
       # and compiles the sources the shared test classes import.
-      - run: yarn build
-      - run: yarn test:postgres
+      - if: env.should_run == 'true'
+        run: yarn build
+      - if: env.should_run == 'true'
+        run: yarn test:postgres
 
   bun-redis:
     runs-on: ubuntu-latest

python/bullmq/backend.py

@@ -348,6 +348,74 @@ async def waitForJob(self, block_timeout: float) -> Any:
         Returns the raw marker entry on success, or a falsy value on timeout.
         """
 
+    # ============================================================
+    # Job schedulers (repeatable job factories)
+    # ============================================================
+
+    @abstractmethod
+    async def addJobScheduler(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        template_opts: dict,
+        scheduler_opts: dict,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        """Register/override a scheduler and enqueue its next iteration.
+
+        Returns a ``(job_id, delay)`` pair for the newly-scheduled iteration,
+        or a falsy value when no iteration was produced.
+        """
+
+    @abstractmethod
+    async def updateJobSchedulerNextMillis(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        """Advance an existing scheduler to its next iteration without
+        touching its template. Returns the new delayed job id, or a falsy
+        value if no iteration was produced (e.g. the scheduler is gone)."""
+
+    @abstractmethod
+    async def removeJobScheduler(self, job_scheduler_id: str) -> int:
+        """Remove a scheduler and its pending next-iteration job.
+
+        Returns 0 if the scheduler was removed, 1 if it did not exist.
+        """
+
+    @abstractmethod
+    async def isJobScheduler(self, job_scheduler_id: str) -> bool:
+        """Return whether ``job_scheduler_id`` is a registered scheduler."""
+
+    @abstractmethod
+    async def getJobScheduler(self, job_scheduler_id: str):
+        """Return a ``(fields, next_millis)`` pair for a single scheduler.
+
+        ``fields`` is the metadata mapping (``name``, ``ic``, ``every``,
+        ``pattern``, ``data``, ``opts`` ...) in the Redis-hash shape that
+        :func:`bullmq.job_scheduler._transform_scheduler_data` consumes, or a
+        falsy value when the scheduler is missing. ``next_millis`` is the
+        next-run timestamp, or ``None``.
+        """
+
+    @abstractmethod
+    async def getJobSchedulers(
+        self, start: int = 0, end: int = -1, asc: bool = False
+    ) -> list:
+        """Return a page of registered schedulers as a list of
+        ``(key, fields, next_millis)`` tuples (see :meth:`getJobScheduler`
+        for the ``fields`` shape)."""
+
+    @abstractmethod
+    async def getJobSchedulersCount(self) -> int:
+        """Return the total number of registered schedulers."""
+
 
 BackendFactory = Callable[..., Backend]
 """Factory that builds a :class:`Backend` for a given queue.

python/bullmq/backends/postgres_backend.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import time
 from typing import Any, Optional, TYPE_CHECKING
@@ -165,6 +166,48 @@ def _normalize_keep(remove_on: Any) -> tuple[bool, Optional[int], Optional[int]]
     return (False, None, None)
 
 
+def _scheduler_row_to_hash(row: dict) -> tuple[dict, Optional[int]]:
+    """Map a ``scheduler`` row into the Redis-hash-shaped ``(fields, next)``
+    pair that :func:`bullmq.job_scheduler._transform_scheduler_data` consumes.
+
+    Scalar fields become string values; ``template_data``/``template_opts``
+    (stored as ``jsonb``, decoded by psycopg into Python objects) are
+    re-serialized to JSON strings and omitted when empty. Absent fields are
+    dropped. Mirrors the Node ``mapSchedulerRow`` helper.
+    """
+    fields: dict = {}
+
+    def _put(key: str, value: Any) -> None:
+        if value is not None:
+            fields[key] = str(value)
+
+    _put("name", row.get("name"))
+    _put("ic", row.get("iteration_count"))
+    _put("limit", row.get("limit_count"))
+    _put("startDate", row.get("start_date_ms"))
+    _put("endDate", row.get("end_date_ms"))
+    _put("tz", row.get("tz"))
+    _put("pattern", row.get("pattern"))
+    _put("every", row.get("every_ms"))
+    _put("offset", row.get("offset_ms"))
+
+    template_data = row.get("template_data")
+    if template_data is not None:
+        data = template_data if isinstance(template_data, str) else _json(template_data)
+        if data != "{}":
+            fields["data"] = data
+
+    template_opts = row.get("template_opts")
+    if template_opts is not None:
+        opts = template_opts if isinstance(template_opts, str) else _json(template_opts)
+        if opts != "{}":
+            fields["opts"] = opts
+
+    next_run = row.get("next_run_ms")
+    next_millis = None if next_run is None else _to_int(next_run)
+    return (fields, next_millis)
+
+
 class PostgresBackend(Backend):
     """PostgreSQL adapter implementing :class:`~bullmq.backend.Backend`."""
 
@@ -753,6 +796,122 @@ async def waitForJob(self, block_timeout: float) -> Any:
             if not checked_waiting_job and await self._has_waiting_job():
                 return marker
 
+    # ============================================================
+    # Job schedulers (repeatable job factories)
+    # ============================================================
+
+    async def addJobScheduler(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        template_opts: dict,
+        scheduler_opts: dict,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        result = await self._run(
+            "add_job_scheduler",
+            [
+                self.queue_name,
+                job_scheduler_id,
+                next_millis,
+                template_data or "{}",
+                _jsonb(template_opts or {}),
+                _jsonb(scheduler_opts or {}),
+                _jsonb(delayed_job_opts or {}),
+                _now_ms(),
+                producer_id,
+            ],
+            op="addJobScheduler",
+        )
+        row = result.first_map()
+        if not row or row.get("job_id") is None:
+            return None
+        return (str(row["job_id"]), _to_int(row.get("delay")))
+
+    async def updateJobSchedulerNextMillis(
+        self,
+        job_scheduler_id: str,
+        next_millis: Optional[int],
+        template_data: str,
+        delayed_job_opts: dict,
+        producer_id: Optional[str] = None,
+    ):
+        result = await self._run(
+            "update_job_scheduler",
+            [
+                self.queue_name,
+                job_scheduler_id,
+                next_millis,
+                template_data or "{}",
+                _jsonb(delayed_job_opts or {}),
+                _now_ms(),
+                producer_id,
+            ],
+            op="updateJobSchedulerNextMillis",
+        )
+        row = result.first_map()
+        job_id = row.get("job_id") if row else None
+        return str(job_id) if job_id is not None else None
+
+    async def removeJobScheduler(self, job_scheduler_id: str) -> int:
+        result = await self._run(
+            "remove_job_scheduler", [self.queue_name, job_scheduler_id]
+        )
+        row = result.first_map()
+        return _to_int(row.get("removed")) if row else 1
+
+    async def isJobScheduler(self, job_scheduler_id: str) -> bool:
+        result = await self._run(
+            "is_job_scheduler", [self.queue_name, job_scheduler_id]
+        )
+        row = result.first_map()
+        return bool(row.get("exists")) if row else False
+
+    async def getJobScheduler(self, job_scheduler_id: str):
+        result = await self._run(
+            "get_job_scheduler", [self.queue_name, job_scheduler_id]
+        )
+        row = result.first_map()
+        if not row:
+            return (None, None)
+        return _scheduler_row_to_hash(row)
+
+    async def getJobSchedulers(
+        self, start: int = 0, end: int = -1, asc: bool = False
+    ) -> list:
+        count = None if end < 0 else end - start + 1
+        result = await self._run(
+            "get_job_schedulers_range", [self.queue_name, asc, start, count]
+        )
+        # The range command only returns (scheduler_id, next_run_ms); fetch
+        # each scheduler's metadata concurrently, mirroring the Redis path.
+        rows = result.maps()
+        if not rows:
+            return []
+        details = await asyncio.gather(
+            *(
+                self._run("get_job_scheduler", [self.queue_name, row["scheduler_id"]])
+                for row in rows
+            )
+        )
+        out = []
+        for row, detail in zip(rows, details):
+            detail_row = detail.first_map()
+            key = row["scheduler_id"]
+            if detail_row:
+                fields, next_millis = _scheduler_row_to_hash(detail_row)
+            else:
+                fields, next_millis = {}, _to_int(row.get("next_run_ms"))
+            out.append((key, fields, next_millis))
+        return out
+
+    async def getJobSchedulersCount(self) -> int:
+        result = await self._run("get_job_schedulers_count", [self.queue_name])
+        row = result.first_map()
+        return _to_int(row.get("count")) if row else 0
+
 
 def create_postgres_backend(
     name: str,