Skip to main content

Python async api for creating and managing queues in postgres

Project description

Python async api for creating and managing queues in postgres

Travis CI PyPI version Documentation Status codecov Code style: black

Postgres is not best solution for storing and managing queues, but sometimes we have no choice.

async-pq can work with millions of entities and thousands of requests in one queue. Is used in production. So its well tested and stable.

Install

> pip install async-pq

Quick start

To work with async-pq we need asyncpg library:

import asyncpg

conn = await asyncpg.connect('postgresql://postgres@localhost/test')

QueueFabric.find_queue method will create needed tables (one for requests and one for messages) in database if it is new queue. Also it has is_exists_queue method for situations when you need to know that it is exists or not and will be the new queue.

from async_pq import Queue, QueueFabric

queue: Queue = await QueueFabric(conn).find_queue('items')

Operations with queue

Put new items (should be dumped JSONs) in queue:

await queue.put('{"id":1,"data":[1,2,3]}', '{"id":2,"data":[3,2,6]}')

Pop items from queue with some limit. It will create one request and return its id. It is useful when you want to use acknowledgement pattern:

# If with_ack=False (default from > 0.2.1), massage will be acknowledged in place automatically
request_id, data = await queue.pop(limit=2, with_ack=True)

To acknowledge request use ack method:

# returns False if request does not found or acked already
is_acked: bool = await queue.ack(request_id)

Or vice versa:

# returns False if request does not found or acked already
is_unacked: bool = await queue.unack(request_id)

You can return unacknowledged massages older than timeout seconds (default limit=1000 requests) to queue:

requests_number = await queue.return_unacked(timeout=300, limit=500)

Clean queue (delete acknowledged massages) to not overfill database with old data (default limit= messages of 1000 requests):

requests_number = await queue.clean_acked_queue(limit=500)

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

async-pq-0.3.1.tar.gz (3.9 kB view hashes)

Uploaded Source

Built Distribution

async_pq-0.3.1-py3-none-any.whl (4.3 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page