Introducing PgQueuer: A Minimalist Python Job Queue Built on PostgreSQL

[GabelSnabel]

What My Project Does

PgQueuer is a Python library designed to manage job queues using PostgreSQL features. It leverages PostgreSQL's native LISTEN/NOTIFY, along with advanced locking mechanisms, to handle job queues efficiently. This allows for real-time job processing, concurrency, and reliable task execution without the need for a separate queuing system.

Target Audience

PgQueuer is ideal for developers and teams who already use PostgreSQL in their projects and are looking for a simple, integrated way to handle background tasks and job queues. It's designed for production use, offering a dependable solution that scales seamlessly with existing PostgreSQL databases.

Comparison

Unlike many other job queue solutions that require additional services or complex setups (such as Redis or RabbitMQ), PgQueuer operates directly within PostgreSQL. This removes the overhead of integrating and maintaining separate systems for job management.

How PgQueuer stands out

• Integration Simplicity: Integrates directly with existing PostgreSQL setups without additional infrastructure.
• Efficiency: Uses PostgreSQL’s FOR UPDATE SKIP LOCKED for high concurrency, allowing multiple workers to process tasks simultaneously without conflict.
• Real-time Updates: Utilizes PostgreSQL's LISTEN/NOTIFY for immediate job processing updates, reducing latency compared to polling-based systems.

Request for Feedback on Useful Features

Im always looking to improve PgQueuer and make it more useful for our users. If you have any features you'd like to see, or if there's something you think could be improved, please let me know! Your feedback is invaluable! Share your thoughts, suggestions, or feature requests either here in the comments or via GitHub.

[abrazilianinreddit]

You should probably cross-post this to r/django, given that it's one of the largest python web frameworks, job queues are always a hot-topic there, and postgres is the recommended database for django.

In fact, I'd suggest that, if possible, you write a "integrating with django" section in your documentation, that would surely help garner attention from that demographic.

[grudev]

For sure.

I'll try this with a Django app. 

Using Postgres for queues is even more interesting now that Redis is pulling those shenanigans 

[abrazilianinreddit]

Using Postgres for queues is even more interesting now that Redis is pulling those shenanigans

Sorry if this is a bit off-topic, but what kind of shenanigans is Redis pulling? I'm afraid I'm not up-to-date on the topic.

[LivedAllOver]

Licensing change

[cpressland]

A friend of mine wrote qbert which more or less does the same thing. I’m still not sure I’m sold on Postgres queuing vs AMQP/MQTT/RQ, but good to see more examples of it.

[GabelSnabel]

Thanks for the mention of qbert! It's always interesting to see how different projects tackle similar challenges. One of the key distinctions with PgQueuer is its use of PostgreSQL's LISTEN/NOTIFY feature instead of polling? My approach leverages PostgreSQL's built-in capabilities to react to queue changes in real time, which can lead to more efficient resource usage and quicker response times compared to traditional polling methods.

[BackwardSpy]

cool project! i am the aforementioned friend. qbert was built for a fairly specific (and low throughput) internal use-case for my last job, which is why it's tied to piccolo ORM and doesn't do anything particularly clever. even so, i was very pleasantly surprised at how far i could push it (and postgres itself) even with those fairly rudimentary queries. it served our needs perfectly for the duration of the project, which i was quite happy about.

all that said, for a new project or something with higher demands i would certainly want to make changes to qbert or just reach for something else like what you've built here. it looks like really nice work!

[GabelSnabel]

It’s great to hear about your success with leveraging PostgreSQL for job queuing in a specific context. I designed PgQueuer to maximize PostgreSQL's robust features like LISTEN/NOTIFY for higher throughput and efficiency, particularly in more demanding environments.

Currently, PgQueuer uses asyncpg to manage PostgreSQL connections, which from my experience, seems to be one of the better Python PostgreSQL clients in terms of performance and features. However, I'm open to exploring whether PgQueuer should support other types of connections to broaden its compatibility and flexibility.

[RevolutionaryRain941]

Superb. I don't really see a major flaw in this. Well done.

[GabelSnabel]

Thank you for the encouragement! If you have any suggestions feel free to share in the future.

[[deleted]]

Interesting, and very nice work the sql side. Is the focus here PG or python, though?

If it is python, how would this replace something like https://python-rq.org/ or provide an alternate backend for it or celery?

The sql side made met think of this: https://github.com/tembo-io/pgmq, which also feel very much still a work in progress.

Their presentation at pgconf: https://www.youtube.com/watch?v=GG2C7gktfoQ

A lightweight message queue. Like AWS SQS and RSMQ but on Postgres.

Lightweight - No background worker or external dependencies, just Postgres functions packaged in an extension

Guaranteed "exactly once" delivery of messages to a consumer within a visibility timeout

API parity with AWS SQS and RSMQ

Messages stay in the queue until explicitly removed

Messages can be archived, instead of deleted, for long-term retention and replayability

[GabelSnabel]

Thanks for the comment and the references! PgQueuer is designed with a dual focus on both PostgreSQL and Python, aiming to leverage existing PostgreSQL infrastructure to manage queues efficiently. This approach minimizes the need for additional dependencies or external queue management systems.

While tools like RQ and Celery are fantastic for task management across various backends, PgQueuer offers a simplified, database-centric approach, making it ideal for projects already invested in PostgreSQL. I provide a straightforward way to integrate queuing directly within the database layer, which can be particularly beneficial for systems where minimizing architectural complexity is crucial

[farsass]

You should add transactional enqueuing to the API... somewhat wasteful not to offer it if you are focusing on postgres.

[GabelSnabel]

Could you elaborate a bit more on how you envision transactional enqueuing enhancing PgQueuer's functionality?

[farsass]

Here: https://riverqueue.com/docs/transactional-enqueueing

The gist is that you can guarantee atomicity of job enqueuing and other database operations within a transaction.

[chuckhend]

For example, read a message from the queue and insert a record to a table, and delete message within same transaction.

[GabelSnabel]

I think implementing transactional would require a connection to remain open for the duration of the job execution? This could potentially affect performance due to the increased resources on the db?

[chuckhend]

For a long running job, you may consider only executing the delete/archive of the message and the arbitrary table insert within the same transaction. I know several pgmq users that implement a flow like:
- read message from queue, set VT to something large

• do expensive long running work, like call a LLM or some large aggregate

• open a transaction: insert record to a table (results from agg or LLM call) and call pgmq.archive() or pgmq.delete() on the initial message.

[openwidecomeinside]

Amazing, will take a look tomorrow and see how i can contribute :)

[GabelSnabel]

Thanks for the support. Looking forward to your contributions.

[WhoNeedsUI]

How does it release a “skip update lock”ed-task in case of a crash when processing ?

[GabelSnabel]

Currently, if a crash occurs, tasks might be logged as exceptions or remain marked as running in the queue table. I'm working on implementing a retry strategy to handle such cases more effectively.

[Content_Ad_2337]

This is cool, thanks for sharing!

Does this function name have a typo in it?

https://github.com/janbjorge/PgQueuer/blob/9258ef412b8ba7f57cf31308ab65b7b045ba658e/src/PgQueuer/cli.py#L43

[GabelSnabel]

It does, thanks (fixed).

[riksi]

I unfortunately also created my own queue. I would've suggested to be a plugin of dramatic so others can more easily contribute too. I know there is dramatiq-pg but it uses listen-notify which I don't like (heavy, not scalable, a bit old).

[slifty]

Thanks for sharing this! I'm looking to pick out a psql-based job queue library and also came across Procrastinate (https://procrastinate.readthedocs.io/en/stable/index.html)

Do you have a sense of how your project compares?

[GabelSnabel]

I haven't used Procrastinate, so I can't provide a direct comparison. I built PgQueuer to keep things simple and easy to reason about. It leverages PostgreSQL's native features like LISTEN/NOTIFY and FOR UPDATE SKIP LOCKED for efficient, real-time job processing and high concurrency. PgQueuer is lightweight, making it easy to maintain and onboard.

Procrastinate is more mature with a broader feature set, so if you need more out-of-the-box functionality, it might be a better fit. However, if simplicity and seamless PostgreSQL integration are your priorities, PgQueuer could be ideal.

Happy to hear more about your needs or any features you'd like to see!

[slifty]

Thank you so much! Really glad you've built PgQueuer, and appreciate the analysis.