Is pgqueuer a good tool to use with web framework like litestar

[takeda]

I plan to use pgqueuer as a mechanism to execute background tasks.

The consumer likely would still run as a task in the main application, is this a suitable use case?

There are currently three items that I'm concerned and I think pgqueuer might not be handling, but perhaps there are workarounds?

• if during execution an exception is thrown I don't want to lose the job. I see that AWS SQS works that makes the job temporarily invisible and once job is completed a signal is sent to mark it as done, otherwise it reappears after specified timeout
• I see the way the code is initialized it takes a single connection, then wraps it in driver and PGQueuer object, but how the code will behave if the connection get lost, can PGQueuer work with a pool?
• log/statistics, my concern is that over time those could grow and fill the database, are they periodically truncated? Or do I need to do it myself? If so are there any precautions about it? (like will wiping them affect functionality in any way?) Can they be turned off?

[janbjorge]

Hi there, thanks for the follow-up! You can embed pgqueuer in your main app to drive background tasks, but it’s worth considering resource isolation: if your queue work suddenly spikes, it could consume CPUs or I/O and starve your primary application threads or event loop. To mitigate this you can:

• Cap concurrency at the entrypoint level using concurrency_limit on the decorator. For example:

  @pgq.entrypoint("my_task", concurrency_limit=5)
  async def my_task(job):
      …

This ensures no more than five instances run in parallel . .

• Bound total workers via the max_concurrent_tasks parameter on PgQueuer.run():

  await PgQueuer(pool_driver).run(max_concurrent_tasks=20)

which caps simultaneous tasks across all entrypoints .

---

1) Exception & retry behavior

By default, thrown exceptions are caught, logged with status "exception" in your log table, and the job is removed from the live queue . To get SQS-style retries:

• retry_timer decorator option

  @pgq.entrypoint("my_task", retry_timer=timedelta(seconds=30))

so any “picked” job whose heartbeat isn’t refreshed within 30 s becomes available again.

• Retry executor
  Swap in RetryWithBackoffEntrypointExecutor via executor= for controlled backoff and retry attempts before final failure .

---

2) Connection handling & pooling

pgqueuer works over any Driver. To survive dropped sessions or share a pool:

• Use AsyncpgPoolDriver wrapping an asyncpg.Pool – it acquires/releases connections and auto-reconnects as needed .
• You can still use a single connection driver if you don’t need pooling, but for production workloads a pool is generally safer.

---

3) Log/statistics table growth

By default pgqueuer appends to queue_table_log and statistics_table but doesn’t auto-truncate. You can manually clean up via:

await queries.clear_queue_log(entrypoint="my_task")
await queries.clear_statistics_log(last=timedelta(days=7))

These methods only touch the log tables and won’t interrupt live queue processing. One option is to add a cron job in pgqueuer that will clean the jobs every 24hours or so.

---

4) CLI command for cleanup (todo?)

I could expose a one-off cleanup CLI(meybe something like this?) If it would help?

@app.command("cleanup-logs")
def cleanup_logs(
    days: int = typer.Option(7, "--days", help="Entries older than this"),
    entrypoints: list[str] = typer.Option(None, "-e", help="Entrypoints to target"),
):
    async def run():
        async with yield_queries(ctx, DBSettings()) as q:
            cutoff = timedelta(days=days)
            await q.clear_statistics_log(entrypoints)
            await q.clear_queue_log(entrypoints)
    asyncio_run(run())

[takeda]

Thank you, this is great.

Though, a lot of those things I couldn't find in the documentation, it would be great to include them. I did saw retry_time mentioned but there wasn't more details how it operates and what exactly is "unprocessed job".

I still think though that the SQS model is safer. I mentioned an exception, but if for example something kills the application then retry time is unlikely to save it.

[janbjorge]

Thanks for the feedback – pgqueuer’s docs could definitely use some love. I’m actively trying to find a technical writer to help expand and clarify the documentation. Below is a draft of a Retry & Visibility section that might go into the docs:

---

How retry_timer works in pgqueuer

When an entrypoint is registered with a retry_timer: timedelta:

@pgq.entrypoint(
  "my_task",
  retry_timer=timedelta(seconds=30),
)
async def my_task(job):
    …

pgqueuer’s dequeue logic combines two streams in one SQL query:

1. New jobs (status = 'queued')
2. Stalled jobs (status = 'picked' whose heartbeat is older than NOW() – retry_timer)

A simplified CTE for picking stalled jobs looks like:

WITH next_job_retry AS (
  SELECT id
    FROM queue_table
   WHERE status = 'picked'
     AND heartbeat < NOW() - retry_after
     AND retry_after > interval '0'
   FOR UPDATE SKIP LOCKED
   LIMIT $1
)
UPDATE queue_table
   SET status           = 'picked',
       heartbeat        = NOW(),
       queue_manager_id = $6
 WHERE id IN (SELECT id FROM next_job_retry)
RETURNING *;

Here, retry_after is exactly your retry_timer value, so any job whose lease has expired becomes eligible for reprocessing.

---

Defining an “unprocessed” (stalled) job

A job is considered unprocessed or stalled if it:

• Was fetched into status = 'picked'
• Has not been marked success or exception
• Has a heartbeat timestamp older than NOW() - retry_timer

This covers both cases where the handler threw (and was logged separately) and where the process was killed (e.g. OOM, SIGKILL) before updating the job’s status.

---

Crash recovery vs. SQS visibility timeout

• SQS leases live outside your app: if the worker dies, the message reappears automatically after the visibility timeout.
• pgqueuer uses Postgres-based leases: dequeue sets status='picked' and stamps heartbeat = NOW(). If the app crashes after that but before marking the job done, the stale heartbeat triggers a retry after your retry_timer interval.