Quick Start

This guide walks you through creating a job consumer, enqueuing work, and monitoring the queue, all in under 5 minutes.

Prerequisites: PgQueuer installed and schema created. See Installation if you haven't done that yet.

---

1. Define a consumer

Create a file called myapp.py. PgQueuer uses a factory pattern: you write an @asynccontextmanager function that yields a configured PgQueuer instance.

=== "asyncpg"

```python
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("fetch")
    async def process_message(job: Job) -> None:
        print(f"Processed: {job!r}")

    @pgq.schedule("cleanup", "0 * * * *")
    async def cleanup(schedule: Schedule) -> None:
        print(f"Hourly cleanup: {datetime.now()!r}")

    yield pgq
```

=== "asyncpg pool"

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

@asynccontextmanager
async def main():
    pool = await asyncpg.create_pool()
    pgq = PgQueuer.from_asyncpg_pool(pool)

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

    @pgq.schedule("cleanup", "0 * * * *")
    async def cleanup(schedule: Schedule) -> None:
        print(f"Hourly cleanup: {datetime.now()!r}")

    yield pgq
```

=== "psycopg"

```python
from contextlib import asynccontextmanager
from datetime import datetime
import psycopg
from pgqueuer import PgQueuer
from pgqueuer.models import Job, Schedule

@asynccontextmanager
async def main():
    connection = await psycopg.AsyncConnection.connect(autocommit=True)
    pgq = PgQueuer.from_psycopg_connection(connection)

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

    @pgq.schedule("cleanup", "0 * * * *")
    async def cleanup(schedule: Schedule) -> None:
        print(f"Hourly cleanup: {datetime.now()!r}")

    yield pgq
```

=== "In-Memory (testing)"

```python
from pgqueuer import PgQueuer
from pgqueuer.models import Job
from pgqueuer.types import QueueExecutionMode

async def test_job_handler():
    pq = PgQueuer.in_memory()

    @pq.entrypoint("say_hello")
    async def say_hello(job: Job) -> None:
        print(f"Processing: {job.payload}")

    await pq.qm.queries.enqueue(["say_hello"], [b"hello"], [0])
    await pq.qm.run(mode=QueueExecutionMode.drain)
```

!!! note psycopg connections must use autocommit=True. PgQueuer relies on LISTEN and immediate visibility of committed rows, which requires autocommit mode.

What's happening here:

• PgQueuer.from_asyncpg_connection(connection) wraps the connection in an AsyncpgDriver and sets up a QueueManager and SchedulerManager.
• @pgq.entrypoint("fetch") registers an async function to handle jobs with entrypoint name "fetch".
• @pgq.schedule("cleanup", "0 * * * *") registers a cron task that runs every hour.
• Yielding the PgQueuer instance tells the CLI how to start your application.

---

2. Run the consumer

pgq run myapp:main

The run command:

1. Imports myapp and enters the main() context manager to get the PgQueuer instance.
2. Registers signal handlers for graceful shutdown (SIGTERM, SIGINT).
3. Starts the QueueManager (job processing) and SchedulerManager (cron tasks) concurrently.
4. Listens on the ch_pgqueuer NOTIFY channel for new work.
5. Shuts down cleanly on Ctrl+C, waiting for in-flight jobs to finish.

---

3. Enqueue jobs

From another process or script, push jobs into the queue:

=== "Async"

```python
import asyncpg
from pgqueuer.db import AsyncpgDriver
from pgqueuer.queries import Queries

async def enqueue_jobs():
    conn = await asyncpg.connect()
    driver = AsyncpgDriver(conn)
    queries = Queries(driver)
    job_ids = await queries.enqueue(
        ["fetch"] * 10,        # entrypoint names
        [b"payload"] * 10,     # payloads (bytes)
        [0] * 10,              # priorities (higher = first)
    )
    print(f"Enqueued {len(job_ids)} jobs")
```

=== "Sync"

```python
import psycopg
from pgqueuer.db import SyncPsycopgDriver
from pgqueuer.queries import SyncQueries

conn = psycopg.connect(autocommit=True)
driver = SyncPsycopgDriver(conn)
queries = SyncQueries(driver)
job_ids = queries.enqueue("fetch", b"payload")
```

=== "CLI"

```bash
pgq queue fetch '{"key": "value"}'
```

Queries.enqueue() accepts lists for batch enqueuing. Each list element corresponds to one job. The method returns the IDs of the newly created jobs. When deduplicating with dedupe_key and on_conflict="skip", the result keeps one entry per input; None marks positions skipped as duplicates. See Reliability & Failure Handling.

---

4. Monitor

Watch your queue in real time:

pgq dashboard --interval 5

This refreshes every 5 seconds and shows jobs per entrypoint, status counts, and throughput.

---

What's next?

You now have a working PgQueuer setup. Here's where to go from here:

Goal	Page
Understand the mental model	Core Concepts
Add concurrency caps	Concurrency Control
Set up cron-style recurring tasks	Scheduling
Handle transient failures with retries	Database-Level Retry
Park failed jobs for human review	Holding Failed Jobs
Wait for a job to finish from another process	Completion Tracking
Deploy workers to production	Deployment
Test without PostgreSQL	In-Memory Adapter
Migrate from a 0.x release	Upgrading from 0.x