Skip to main content

Python async api for creating and managing queues in postgres

Project description

Travis CI PyPI version Documentation Status codecov Code style: black


> pip install async-pq


To work with queue we need asyncpg library

import asyncpg

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

QueueFabric.find_queue method will create needed tables in database if it is new queue. Also it has is_exists_queue method for situations when you need to know that it will be the new queue

from async_pq import Queue, QueueFabric

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

Put new items (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 is possible to use acknowledge pattern

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

Acknowledge request

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

Return to queue all unacknowledged massages older than timeout seconds

await queue.return_unacked(timeout=300)

Clean queue (delete all acknowledged massages)

await queue.clean_acked_queue()

Project details

Download files

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

Filename, size & hash SHA256 hash help File type Python version Upload date
async_pq-0.2.1-py3-none-any.whl (3.7 kB) Copy SHA256 hash SHA256 Wheel py3 Aug 10, 2018
async-pq-0.2.1.tar.gz (3.5 kB) Copy SHA256 hash SHA256 Source None Aug 10, 2018

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page