Feature request: provide a “flush all jobs” API/CLI command

[matecsaj]

I couldn’t find a built-in way to flush all jobs (remove every row in procrastinate_jobs) via the API or CLI.

My first thought was to run a raw SQL command:

Truncate Table procrastinate_jobs

…but I wasn’t sure whether that might leave Procrastinate in an inconsistent state (e.g. other tables, sequences, or bookkeeping).

To be safe, I instead put together a relatively slow helper function using only the official library calls (listing jobs, cancelling/aborting, retrying stalled jobs, then deleting finished ones). It works, but it’s not very efficient compared to a single operation.

👉 Feature request: add a supported way (API method and/or CLI flag) to “flush all jobs,” i.e. remove every job record across all statuses, in a way that guarantees internal consistency.

This would be useful for:
• local development / testing, when you want to start with a clean slate
• CI pipelines that need to reset the state quickly
• avoiding direct SQL in application code

import asyncio
from procrastinate.jobs import Status
from wat_backend.procrastinate_app import app
import logging

async def flush_all_jobs(
    grace_seconds: float = 5.0,
    poll_interval: float = 0.5,
) -> None:
    """
    Flush all Procrastinate jobs using only the Procrastinate API (no direct SQL).

    Args:
        grace_seconds (float): Maximum time in seconds to wait for running workers
            to acknowledge abort requests. Set to 0 to skip waiting.
        poll_interval (float): Interval in seconds between checks while waiting
            for running jobs to stop during the grace period.

    Returns:
        None

    Notes:
        - This function opens and closes the app context internally.
        - It tends to run faster when there are no operational workers connected,
          because no grace-period waiting is required.
        - It logs a warning if any jobs remain in any status afterward. (should never happen)
    """
    await app.open_async()
    try:
        jm = app.job_manager

        # 1) Cancel & delete queued jobs (TO DO)
        todos = await jm.list_jobs_async(status=Status.TODO.value)
        for job in todos:
            await jm.cancel_job_by_id_async(job.id, delete_job=True)

        # 2) Request abort for running jobs (DOING)
        doing = await jm.list_jobs_async(status=Status.DOING.value)
        for job in doing:
            await jm.cancel_job_by_id_async(job.id, abort=True)

        # 3) Grace period: let user's workers acknowledge aborts
        if doing and grace_seconds > 0:
            loop = asyncio.get_running_loop()
            deadline = loop.time() + grace_seconds
            while loop.time() < deadline:
                still_doing = await jm.list_jobs_async(status=Status.DOING.value)
                if not still_doing:
                    break
                await asyncio.sleep(poll_interval)

        # 4) Handle stalled jobs, this can happen when workers don't acknowledge aborts
        #  a work-around is to requeue them and then cancel & delete
        stalled = await jm.get_stalled_jobs()
        for job in stalled:
            await jm.retry_job(job)
        for job in stalled:
            await jm.cancel_job_by_id_async(job.id, delete_job=True)

        # 5) Delete all finished jobs (older than "now" → nb_hours=0)
        await jm.delete_old_jobs(
            nb_hours=0,
            include_failed=True,
            include_cancelled=True,
            include_aborted=True,
        )

        # 6) Final check: are there any jobs left in *any* status?
        # This should never trigger; did this function miss an edge case, or does Procrastinate have a bug?
        total_leftover = 0
        for st in Status:
            jobs = await jm.list_jobs_async(status=st.value)
            total_leftover += len(jobs)

        if total_leftover:
            logging.warning(f"Flush incomplete — {total_leftover} jobs remain")

    finally:
        await app.close_async()

[matecsaj]

After adding a periodic task to my code, I realized the function I shared earlier was incomplete.

# Delete old jobs every day at 4am. This prevents the database from overflowing with completed jobs.
@app.periodic(cron="0 4 * * *")
@app.task(queueing_lock="remove_old_jobs", queue="orchestrate", pass_context=True)
async def remove_old_jobs(context, timestamp):

    for queue in ["extract", "transform", "load"]:
        result = await builtin_tasks.remove_old_jobs(
            context,
            max_hours=72,
            queue=queue,
            remove_failed=True,
            remove_cancelled=True,
            remove_aborted=True,
        )
    return result

Since I couldn’t find a Procrastinate API call that deletes periodic tasks from the database, I had to fall back on a raw SQL workaround:

async def flush_periodic_defers() -> None:
    """
    Danger: Flush Procrastinate's periodic deferral ledger.

    This deletes *all rows* from the `procrastinate_periodic_defers` table,
    which is Procrastinate's internal bookkeeping to ensure that each
    @app.periodic task is only deferred once per schedule.

    Why we resort to raw SQL:
    -------------------------
    - Procrastinate exposes no public API for flushing this table.
    - For development/testing it is sometimes convenient to "start fresh"
      and force periodic tasks (e.g. a monthly orchestrator) to be enqueued again.
    - The only reliable way to do this is executing
        DELETE FROM public.procrastinate_periodic_defers;

    Why this is a bad idea:
    -----------------------
    - You are manipulating Procrastinate's internal state directly, bypassing
      its public library calls.
    - Doing this in production could break invariants or cause duplicate
      scheduling of periodic tasks.
    - There is no guarantee future versions of Procrastinate will use the same
      schema or table names.

    Use this only in development or CI where you explicitly need to re-run
    periodic tasks as if they had never been scheduled.
    """
    await app.open_async()
    try:
        await app.connector.execute_query_async(
            "DELETE FROM public.procrastinate_periodic_defers;"
        )
    finally:
        await app.close_async()

[matecsaj]

In case you’re wondering why I didn’t follow the advice in Periodic task lifecycle management when writing flush_periodic_defers: I couldn’t find a Procrastinate call that returns the list of periodic jobs.