Scheduling

PgQueuer includes a built-in scheduler for managing recurring tasks using cron-like expressions. No separate process (like celery-beat) is required.

Basic usage

from pgqueuer.models import Schedule

@pgq.schedule("fetch_db", "* * * * *")
async def fetch_db(schedule: Schedule) -> None:
    await perform_task()

Use a 6-field expression to schedule work at second-level granularity:

@pgq.schedule("heartbeat", "* * * * * */3")
async def heartbeat(schedule: Schedule) -> None:
    await send_heartbeat()

How it works

The @schedule decorator registers a name and a cron expression as a row in pgqueuer_schedules. The scheduler polls that table, runs the tasks whose expressions are due, and records the run. Because the state lives in PostgreSQL, schedules survive process restarts and only one worker runs each due task.

Scheduler flow diagram

@pgq.schedule ──▶ Schedule in DB ──▶ Poll loop ──cron ready──▶ Execute task
                                        ▲                          │
                                        └──────────────────────────┘

Cron expression format

PgQueuer supports both standard 5-field cron expressions and 6-field expressions with seconds in the final position.

5-field format

┌───────────── minute (0-59)
│ ┌───────────── hour (0-23)
│ │ ┌───────────── day of month (1-31)
│ │ │ ┌───────────── month (1-12)
│ │ │ │ ┌───────────── day of week (0-6, Sunday=0)
│ │ │ │ │
* * * * *

Examples:

Expression	Meaning
* * * * *	Every minute
*/5 * * * *	Every 5 minutes
0 * * * *	Every hour
0 9 * * 1	Every Monday at 9:00 AM
0 0 1 * *	First day of each month at midnight

6-field format

For second-level schedules, the seconds field comes last:

PgQueuer uses croniter to parse cron expressions. Second-level schedules follow croniter's about second repeats behavior.

┌───────────── minute (0-59)
│ ┌───────────── hour (0-23)
│ │ ┌───────────── day of month (1-31)
│ │ │ ┌───────────── month (1-12)
│ │ │ │ ┌───────────── day of week (0-6, Sunday=0)
│ │ │ │ │ ┌───────────── second (0-59)
│ │ │ │ │ │
* * * * * *

Examples:

Expression	Meaning
* * * * * */3	Every 3 seconds
* * * * * */10	Every 10 seconds
* * * * * 0	At second 0 of every minute

Use trailing seconds, not leading seconds. For example, */3 * * * * * is not "every 3 seconds"; use * * * * * */3 instead.

Accessing shared resources

Scheduled tasks can access shared resources (HTTP clients, database pools, etc.) by annotating a parameter as ScheduleContext. PgQueuer auto-detects it and injects the context, whose .resources mapping mirrors the one passed to PgQueuer:

from pgqueuer.models import Schedule, ScheduleContext

@pgq.schedule("sync_data", "0 * * * *")
async def sync_data(schedule: Schedule, ctx: ScheduleContext) -> None:
    db = ctx.resources["db_pool"]
    await db.execute("SELECT 1")

See Shared Resources for full details.

Cleaning up old schedules

When clean_old=True, PgQueuer deletes any existing database schedules whose entrypoint matches this decorator's entrypoint before re-registering it. This is useful when you change the cron expression for a task and want the old schedule row replaced on startup.

@pgq.schedule("fetch_db", "* * * * *", clean_old=True)
async def fetch_db(schedule: Schedule) -> None:
    await perform_task()

By default, clean_old=False: old schedules are retained.

Managing schedules via CLI

List all schedules:

pgq schedules

Remove a schedule by name:

pgq schedules --remove fetch_db

Example: full setup with scheduling

from contextlib import asynccontextmanager
from datetime import datetime
import asyncpg
from pgqueuer import PgQueuer
from pgqueuer.models import Job, Schedule

@asynccontextmanager
async def main():
    connection = await asyncpg.connect()
    pgq = PgQueuer.from_asyncpg_connection(connection)

    @pgq.entrypoint("process_job")
    async def process_job(job: Job) -> None:
        print(f"Processing: {job!r}")

    @pgq.schedule("hourly_report", "0 * * * *")
    async def hourly_report(schedule: Schedule) -> None:
        print(f"Generating report at {datetime.now()}")

    @pgq.schedule("heartbeat", "* * * * * */3")
    async def heartbeat(schedule: Schedule) -> None:
        print(f"Heartbeat at {datetime.now()}")

    @pgq.schedule("daily_cleanup", "0 2 * * *", clean_old=True)
    async def daily_cleanup(schedule: Schedule) -> None:
        print(f"Running cleanup at {datetime.now()}")

    yield pgq

pgq run myapp:main