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