PgQueuer

https://janbjorge.github.io/pgqueuer
Admin
PgQueuer is a PostgreSQL-backed job queue for Python that provides fast and reliable background job...
Tokens:70,736
Snippets:705
Trust Score:9.7
Update:1 month ago
Show doc for...
Tokens:
Raw
### Quick start: install dependencies and run checks

Source: https://janbjorge.github.io/pgqueuer/development/development

Quick start commands to set up the environment, run linting, type checks, and the full test suite. Requires Python 3.10+ and a Docker daemon; the test suite automatically starts and tears down a disposable PostgreSQL instance via Testcontainers.

```bash
# 1. Install dependencies (including all extras)
uv sync --all-extras --frozen

# 2. Lint and formatting checks
uv run ruff check .

# 3. Type checks
uv run mypy .

# 4. Full test suite (auto-starts and tears down a disposable PostgreSQL)
uv run pytest

```

--------------------------------

### Full scheduling setup with PgQueuer

Source: https://janbjorge.github.io/pgqueuer/guides/scheduling

Complete example of an asynccontextmanager that creates an asyncpg connection, initializes PgQueuer, and registers an entrypoint along with several scheduled tasks (hourly_report, heartbeat, daily_cleanup). It yields the pgq instance for use by a runner.

```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("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


```

--------------------------------

### Install PgQueuer with multiple extras

Source: https://janbjorge.github.io/pgqueuer/getting-started/installation

Install PgQueuer with multiple extras at once, here asyncpg and logfire.

```bash
pip install pgqueuer[asyncpg,logfire]

```

--------------------------------

### Create MCP server with custom connection settings and run

Source: https://janbjorge.github.io/pgqueuer/integrations/mcp-server

Creates an MCP server with an explicit DSN and custom `ConnectionSettings` for pool sizing, then runs it with the stdio transport. This example shows the full setup including the import of `ConnectionSettings` and the `server.run()` call.

```python
# Pool sizing/timeouts can also be passed explicitly
from pgqueuer.domain.settings import ConnectionSettings

server = create_mcp_server(
    dsn="postgresql://...",
    connection_settings=ConnectionSettings(pool_min_size=2, pool_max_size=10),
)

server.run(transport="stdio")
```

--------------------------------

### Install pgqueuer[fastapi] optional extra

Source: https://janbjorge.github.io/pgqueuer/integrations/prometheus

Install the optional extra required for FastAPI integration.

```bash
pip install pgqueuer[fastapi]

```

--------------------------------

### Install PgQueuer with asyncpg, psycopg, or uv

Source: https://janbjorge.github.io/pgqueuer/getting-started/installation

Install PgQueuer with the asyncpg driver (recommended), the psycopg driver for sync code, or via uv with asyncpg.

```bash
pip install pgqueuer[asyncpg]

```

```bash
pip install pgqueuer[psycopg]

```

```bash
uv add pgqueuer[asyncpg]

```

--------------------------------

### pgq install --durability balanced

Source: https://janbjorge.github.io/pgqueuer/reference/cli

Installs the PgQueuer database schema with the balanced durability level. On success, a confirmation is written to stderr; stdout stays empty. To preview or capture the SQL instead of executing it, use `pgq sql install`.

```bash
pgq install --durability balanced
```

--------------------------------

### Install PgQueuer with MCP extras (command line)

Source: https://janbjorge.github.io/pgqueuer/integrations/mcp-server

Install pgqueuer with the required MCP and asyncpg dependencies. Use with pip or pip.

```bash
uv add pgqueuer[mcp]  # displayed as code block under 'uv' tab
```

```bash
pip install pgqueuer[mcp]  # displayed as code block under 'pip' tab
```

--------------------------------

### pgq run - Start QueueManager with options

Source: https://janbjorge.github.io/pgqueuer/reference/cli

Starts a QueueManager to process jobs. Supports flags for dequeue timeout, batch size, heartbeat timeout, restart delay, restart on failure, log level, mode (continuous or drain), max concurrent tasks, and shutdown on listener failure. The example shows running with a limit of 5 concurrent tasks and drain mode.

```bash
# Run with a limit of 5 concurrent tasks
pgq run my_module:my_factory --max-concurrent-tasks 5

# Drain mode: process all queued jobs then exit
pgq run my_module:my_factory --mode drain
```

--------------------------------

### Install PgQueuer web dashboard

Source: https://janbjorge.github.io/pgqueuer/integrations/web-dashboard

Install the web dashboard with the asyncpg driver. The extra includes FastAPI, htmx, and the asyncpg pool driver.

```bash
pip install "pgqueuer[web,asyncpg]"
```

--------------------------------

### Quick Start: In-Memory Job Processing

Source: https://janbjorge.github.io/pgqueuer/reference/in-memory

A complete example that sets up an in-memory queue, enqueues five jobs, and processes them in drain mode. Uses QueueExecutionMode.drain, batch_size of 10, and max_concurrent_tasks of 100. Handlers print job processing and completion messages.

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

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

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

    job_ids = await pq.qm.queries.enqueue(
        ["say_hello"] * 5,
        [b"job 1", b"job 2", b"job 3", b"job 4", b"job 5"],
        [0] * 5,
    )
    print(f"Enqueued {len(job_ids)} jobs")

    await pq.qm.run(
        batch_size=10,
        mode=QueueExecutionMode.drain,
        max_concurrent_tasks=100,
    )
    print("All jobs processed!")

asyncio.run(main())
```

--------------------------------

### Preview and apply schema SQL with pgq sql install

Source: https://janbjorge.github.io/pgqueuer/getting-started/installation

Preview the SQL without executing it, and pipe it to psql for applying with error stopping.

```bash
pgq sql install

```

```bash
pgq sql install | psql -v ON_ERROR_STOP=1

```

--------------------------------

### Install and run PgQueuer with a custom schema (pgq --schema)

Source: https://janbjorge.github.io/pgqueuer/getting-started/installation

Installs PgQueuer into a dedicated Postgres schema named 'pgq' and runs the application with that schema. The install command creates the schema if it does not exist; if the role lacks CREATE on the database and the schema already exists, pass --no-create-schema. All tables, the enum type, and the trigger function are created in and referenced through that schema, regardless of the connection's search_path. The NOTIFY channel is unaffected because channels are global to the database.

```shell
pgq --schema pgq install
PGQUEUER_SCHEMA=pgq pgq run my_app:main

```

--------------------------------

### pgq install

Source: https://janbjorge.github.io/pgqueuer/reference/cli

Sets up the necessary database schema for PgQueuer. Use the --durability option to control table logging behavior.

```APIDOC
## [CLI] pgq install

### Description
Set up the necessary database schema for PgQueuer.

### Method
CLI command

### Endpoint
`pgq install`

### Parameters
#### Options
- `--durability` (string) - Optional - Define the durability level for tables. Values: `volatile`, `balanced`, `durable` (default).
- `--dry-run` (deprecated) - Alias for `pgq sql install`.

### Request Example
```
pgq install --durability balanced
```

### Response
On success, a confirmation is written to stderr; stdout stays empty. To preview or capture the SQL instead of executing it, use `pgq sql install`.

```

--------------------------------

### Run the consumer with `pgq run myapp:main`

Source: https://janbjorge.github.io/pgqueuer/getting-started/quickstart

Starts the PgQueuer consumer by importing `myapp` and entering the `main()` context manager. It registers signal handlers for graceful shutdown, starts the `QueueManager` and `SchedulerManager` concurrently, listens on the `ch_pgqueuer` NOTIFY channel, and shuts down cleanly on Ctrl+C.

```bash
pgq run myapp:main

```

--------------------------------

### Install optional tracing dependencies

Source: https://janbjorge.github.io/pgqueuer/integrations/tracing

Install optional dependencies for Logfire, Sentry, or OpenTelemetry tracing. Multiple extras can be installed simultaneously.

```bash
pip install pgqueuer[logfire]
# or
pip install pgqueuer[sentry]
# or
pip install pgqueuer[opentelemetry]

```

--------------------------------

### Install without creating schema

Source: https://janbjorge.github.io/pgqueuer/reference/database-permissions

Command to run `pgq install` with `--no-create-schema` after pre-creating the schema.

```bash
pgq --schema pgq install --no-create-schema

```

--------------------------------

### Installation privileges for pgq install

Source: https://janbjorge.github.io/pgqueuer/reference/database-permissions

Required one-time privileges for the database role that runs `pgq install`. Includes GRANT statements for database or schema creation, and comments listing object-level permissions needed.

```sql
-- Required one-time privileges for schema installation
GRANT CREATE ON DATABASE mydb TO pgqueuer_installer;
-- Or, if using a dedicated schema:
GRANT CREATE ON SCHEMA pgqueuer_schema TO pgqueuer_installer;

-- Specific object-level permissions needed:
-- CREATE TYPE   → for pgqueuer_status ENUM
-- CREATE TABLE  → for all 4 tables
-- CREATE INDEX  → for all indexes
-- CREATE FUNCTION → for fn_pgqueuer_changed()
-- CREATE TRIGGER  → for tg_pgqueuer_changed

```

--------------------------------

### Run the MCP server

Source: https://janbjorge.github.io/pgqueuer/integrations/mcp-server

Start a stdio-based MCP server. Use the appropriate command for your environment. It connects using libpq environment variables.

```bash
uv run python -m pgqueuer.adapters.mcp
```

```bash
python -m pgqueuer.adapters.mcp
```

--------------------------------

### Verify PgQueuer schema installation with pgq verify

Source: https://janbjorge.github.io/pgqueuer/getting-started/installation

Run `pgq verify --expect present` to confirm all PgQueuer objects are present. Exits with code 0 if the schema is correctly installed, or code 1 if anything is missing.

```bash
pgq verify --expect present

```

--------------------------------

### Pre-create schema for locked-down role

Source: https://janbjorge.github.io/pgqueuer/reference/database-permissions

SQL to pre-create the `pgq` schema and grant necessary privileges to installer and application roles, allowing install without schema creation.

```sql
CREATE SCHEMA pgq;
GRANT USAGE, CREATE ON SCHEMA pgq TO pgqueuer_installer;
GRANT USAGE ON SCHEMA pgq TO pgqueuer_app;

```

--------------------------------

### Explicit DSN example

Source: https://janbjorge.github.io/pgqueuer/integrations/mcp-server

If you prefer a connection string over environment variables, use this factory function and run the server.

```python
from pgqueuer.adapters.mcp.server import create_mcp_server

server = create_mcp_server(dsn="postgresql://user:pass@host:5432/mydb")
server.run(transport="stdio")
```

--------------------------------

### PsycopgDriver setup

Source: https://janbjorge.github.io/pgqueuer/reference/drivers

Creates a PsycopgDriver from an async psycopg connection. Requires autocommit=True before handing the connection to the driver.

```python
import psycopg
from pgqueuer.db import PsycopgDriver

conn = await psycopg.AsyncConnection.connect(dsn)
conn.autocommit = True
driver = PsycopgDriver(conn)

```

--------------------------------

### Development flow summary

Source: https://janbjorge.github.io/pgqueuer/development/development

Development flow summary: install dependencies, run linting, import checks, type checks, and the full test suite. This is the recommended sequence for contributing to PgQueuer.

```bash
uv sync --all-extras --frozen
uv run ruff check .
uv run lint-imports
uv run mypy .
uv run pytest -v

```

--------------------------------

### Generate SQL offline with pgq sql install and pgq sql upgrade

Source: https://janbjorge.github.io/pgqueuer/reference/database-setup

Shows the SQL that would be executed without connecting to a database. The first command prints the install script; the second pipes it to psql; the third generates an upgrade migration file. Useful for offline review, migration tools (Flyway, sqitch, Alembic), or a DBA with restricted permissions.

```bash
pgq sql install

```

```bash
pgq sql install | psql -v ON_ERROR_STOP=1

```

```bash
pgq sql upgrade > migrations/V2__pgqueuer_upgrade.sql

```

--------------------------------

### Install PgQueuer with durability balanced

Source: https://janbjorge.github.io/pgqueuer/reference/database-setup

Installs the schema with a different durability level (here 'balanced') instead of the default 'durable'. See CLI Reference for full details on volatile, balanced, and durable modes.

```bash
pgq install --durability balanced

```

--------------------------------

### Build and run PgQueuer dashboard Docker image

Source: https://janbjorge.github.io/pgqueuer/integrations/web-dashboard

Build and run the dashboard Docker image. The Dockerfile is at tools/web/Dockerfile and a compose example is at tools/web/docker-compose.yml.

```bash
docker build -f tools/web/Dockerfile -t pgqueuer-dashboard .
docker run -e PGHOST=db -e PGUSER=... -p 8080:8080 pgqueuer-dashboard
```

--------------------------------

### Create PgQueuer schema with pgq install

Source: https://janbjorge.github.io/pgqueuer/getting-started/installation

Create the PgQueuer database schema (tables, enum, function, trigger) using the CLI.

```bash
pgq install

```

--------------------------------

### PgQueuer.from_asyncpg_connection example

Source: https://janbjorge.github.io/pgqueuer/reference/drivers

Creates a PgQueuer instance from an asyncpg connection, optionally with shared resources, then runs it.

```python
import asyncpg
from pgqueuer import PgQueuer

connection = await asyncpg.connect(dsn)
pgq = PgQueuer.from_asyncpg_connection(connection)

# With optional shared resources
pgq = PgQueuer.from_asyncpg_connection(
    connection,
    resources={"shared_cache": {}},
)

await pgq.run()

```

--------------------------------

### Define consumer and run with pgq CLI

Source: https://janbjorge.github.io/pgqueuer

Defines a consumer with an async context manager, registers a job handler for the 'send_email' entrypoint, and shows the CLI commands to install the schema and run the consumer.

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

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

    @pgq.entrypoint("send_email")
    async def send_email(job: Job) -> None:
        print(f"Sending email: {job.payload}")

    yield pgq

```

```bash
pgq install          # create the schema
pgq run myapp:main   # start processing

```

--------------------------------

### SyncPsycopgDriver enqueue example

Source: https://janbjorge.github.io/pgqueuer/reference/drivers

Sets up the synchronous driver for blocking code or frameworks like Flask. Can only enqueue jobs; consumers require an async driver.

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

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

```

--------------------------------

### Define a consumer with asyncpg connection

Source: https://janbjorge.github.io/pgqueuer/getting-started/quickstart

Creates a PgQueuer instance from an asyncpg connection. Registers an entrypoint 'fetch' and a schedule 'cleanup' that runs hourly. Yields the PgQueuer instance for the CLI to start the application.

```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

```

--------------------------------

### Define a consumer with asyncpg pool

Source: https://janbjorge.github.io/pgqueuer/getting-started/quickstart

Creates a PgQueuer instance from an asyncpg pool. Registers an entrypoint 'fetch' and a schedule 'cleanup' that runs hourly. Yields the PgQueuer instance for the CLI to start the application.

```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

```

--------------------------------

### Install PgQueuer into dedicated schema

Source: https://janbjorge.github.io/pgqueuer/reference/database-permissions

Command to install PgQueuer into its own schema `pgq` using the `--schema` flag.

```bash
pgq --schema pgq install

```

--------------------------------

### CLI connection options: before and after

Source: https://janbjorge.github.io/pgqueuer/getting-started/upgrading

Shows the replacement of per-component connection flags with libpq environment variables. The six flags (--pg-host, --pg-port, --pg-user, --pg-database, --pg-password, --pg-schema) are gone; both asyncpg and psycopg read standard libpq env vars natively when no DSN is provided. The custom schema example uses --schema.

```bash
# Before
pgq --pg-host db.internal --pg-user app --pg-database queue install

# After
PGHOST=db.internal PGUSER=app PGDATABASE=queue pgq install

# Custom schema
pgq --schema myschema install
```

--------------------------------

### Create MCP server with default settings

Source: https://janbjorge.github.io/pgqueuer/integrations/mcp-server

Creates an MCP server using the `create_mcp_server` factory with no arguments. All parameters are optional; the DSN is read from `PGQUEUER_DSN`/`PGDSN` environment variables, otherwise asyncpg reads libpq environment variables. Pool sizing is configured via `PGQUEUER_*` environment variables.

```python
from pgqueuer.adapters.mcp.server import create_mcp_server

# All parameters are optional; the DSN comes from PGQUEUER_DSN/PGDSN,
# else asyncpg reads libpq env vars. Pool sizing from PGQUEUER_* env vars.
server = create_mcp_server()
```

--------------------------------

### Create PgQueuer instances with different drivers

Source: https://janbjorge.github.io/pgqueuer/getting-started/core-concepts

Shows factory classmethods on PgQueuer for setting up with different drivers: asyncpg single connection, asyncpg pool, psycopg async connection, and in-memory (no database). Each method returns a PgQueuer instance configured for the corresponding driver.

```python
# asyncpg single connection
pgq = PgQueuer.from_asyncpg_connection(conn)

# asyncpg pool
pgq = PgQueuer.from_asyncpg_pool(pool)

# psycopg async connection
pgq = PgQueuer.from_psycopg_connection(conn)

# In-memory (no database)
pgq = PgQueuer.in_memory()
```

--------------------------------

### Install and run PgQueuer with a custom table prefix (PGQUEUER_PREFIX)

Source: https://janbjorge.github.io/pgqueuer/getting-started/installation

Installs and runs PgQueuer with a custom table prefix 'billing' to isolate multiple instances in the same database or schema. The prefix applies to all table names, the enum type, the trigger, and the NOTIFY channel. This setting is independent of PGQUEUER_SCHEMA; a prefix separates instances that share a schema, while PGQUEUER_SCHEMA moves the whole installation into another schema.

```shell
PGQUEUER_PREFIX=billing pgq install
PGQUEUER_PREFIX=billing pgq run billing_app:main

```

--------------------------------

### Constructing concrete adapters at the composition root

Source: https://janbjorge.github.io/pgqueuer/reference/ports-and-adapters

Shows how callers construct concrete adapters at the composition root by creating an AsyncpgDriver and passing it to Queries, then to QueueManager.

```python
driver = AsyncpgDriver(connection)
qm = QueueManager(Queries(driver))

```

--------------------------------

### PgQueuer automatic partial indexes

Source: https://janbjorge.github.io/pgqueuer/guides/performance-tuning

These partial indexes are installed automatically by pgq install and pgq upgrade. The first is used for every dequeue (priority-ordered job selection); the second finds stale 'picked' jobs for crash/retry recovery. Do not drop them.

```sql
-- Used for every dequeue: priority-ordered job selection
CREATE INDEX ON pgqueuer (priority ASC, id DESC)
INCLUDE (id) WHERE status = 'queued';

-- Used to find stale 'picked' jobs for crash/retry recovery
CREATE INDEX ON pgqueuer (updated ASC, id DESC)
INCLUDE (id) WHERE status = 'picked';

```

--------------------------------

### Testing with resources

Source: https://janbjorge.github.io/pgqueuer/guides/shared-resources

Create a QueueManager with resources for testing, and assert resource values inside entry points.

```python
from pgqueuer.queries import Queries

qm = QueueManager(Queries(driver), resources={"flag": "test"})

@qm.entrypoint("demo")
async def demo(job: Job, ctx: Context) -> None:
    assert ctx.resources["flag"] == "test"

```

--------------------------------

### Factory contract example: async context manager

Source: https://janbjorge.github.io/pgqueuer/reference/cli

Example of a factory function that must return an AsyncContextManager, typically via @asynccontextmanager. It creates a connection and registers an entrypoint. Passing a bare awaitable or sync context manager raises TypeError.

```python
from contextlib import asynccontextmanager
import asyncpg
from pgqueuer import PgQueuer

@asynccontextmanager
async def create_pgqueuer():
    conn = await asyncpg.connect()
    pgq = PgQueuer.from_asyncpg_connection(conn)

    @pgq.entrypoint("fetch")
    async def process(job): ...

    yield pgq
```

--------------------------------

### Run throughput benchmark with default settings

Source: https://janbjorge.github.io/pgqueuer/comparisons/benchmarks

Runs the default throughput strategy for 30 seconds with 5 dequeue workers and batch size 10. The output shows the settings table and a tqdm progress line; exact numbers vary by environment.

```bash
python3 tools/benchmark.py --dequeue 5 --dequeue-batch-size 10

```

```text
+--------------------+------------+
| Field              | Value      |
+--------------------+------------+
| Driver             | apg        |
| Strategy           | throughput |
| Timer (s)          | 30.0       |
| Dequeue Tasks      | 5          |
| Dequeue Batch Size | 10         |
| Enqueue Tasks      | 1          |
| Enqueue Batch Size | 10         |
| Output JSON        | None       |
+--------------------+------------+
550k job [00:30, 18.3k job/s]

```

--------------------------------

### PgQueuer.from_psycopg_connection example

Source: https://janbjorge.github.io/pgqueuer/reference/drivers

Creates a PgQueuer instance from a psycopg async connection with autocommit enabled.

```python
import psycopg
from pgqueuer import PgQueuer

connection = await psycopg.AsyncConnection.connect(dsn, autocommit=True)
pgq = PgQueuer.from_psycopg_connection(connection)

await pgq.run()

```

--------------------------------

### PgQueuer.from_asyncpg_pool example

Source: https://janbjorge.github.io/pgqueuer/reference/drivers

Creates a PgQueuer instance from an asyncpg pool for improved throughput under high concurrency.

```python
import asyncpg
from pgqueuer import PgQueuer

pool = await asyncpg.create_pool(dsn, min_size=2, max_size=10)
pgq = PgQueuer.from_asyncpg_pool(pool)

await pgq.run()

```

--------------------------------

### Retry metadata structure

Source: https://janbjorge.github.io/pgqueuer/guides/retry

Example of the JSON structure stored in the additional_context field of the pgqueuer_log table for a retry event.

```json
{
    "entrypoint": "call_api",
    "queue_manager_id": "1f4d0c9e-...",
    "attempt": 0,
    "retry_delay": "0:00:30",
    "reason": "rate limited"
}
```

--------------------------------

### Use case diagram: Operate installation includes Process jobs includes Execute job

Source: https://janbjorge.github.io/pgqueuer/design

Use case diagram showing the top-down inclusion of three nested use cases: Operate installation includes Process jobs, which includes Execute job. A fourth use case, Run schedules (UC2b), sits beside Process jobs inside the same worker process.

```text
Operate installation ──includes──► Process jobs ──includes──► Execute job

```

--------------------------------

### Custom table prefix or schema settings

Source: https://janbjorge.github.io/pgqueuer/integrations/mcp-server

Use this example to read custom settings from PGQUEUER_PREFIX or PGQUEUER_SCHEMA for namespaced tables.

```python
from pgqueuer.adapters.mcp.server import create_mcp_server
from pgqueuer.adapters.persistence.qb import DBSettings

# reads PGQUEUER_PREFIX / PGQUEUER_SCHEMA
server = create_mcp_server(settings=DBSettings())
server.run(transport="stdio")
```

--------------------------------

### Migrate manager constructors to RepositoryPort

Source: https://janbjorge.github.io/pgqueuer/getting-started/upgrading

Shows the required migration from passing a raw driver to QueueManager, SchedulerManager, and CompletionWatcher to using a Queries instance. The driver is now accessed via queries.driver, and CompletionWatcher now requires the queries argument.

```python
# Before
qm = QueueManager(driver)
sm = SchedulerManager(driver)
watcher = CompletionWatcher(driver)

# After
from pgqueuer.queries import Queries

queries = Queries(driver)
qm = QueueManager(queries)
sm = SchedulerManager(queries)
watcher = CompletionWatcher(driver, queries=queries)
```

--------------------------------

### Set heartbeat_timeout via CLI

Source: https://janbjorge.github.io/pgqueuer/guides/heartbeat

Command-line equivalent to set the heartbeat timeout to 300 seconds (5 minutes) when starting a worker.

```bash
pgq run my_module:my_factory --heartbeat-timeout 300
```
Tokens:
Raw