pycq

Simple distributed FIFO queue for Python using CockroachDB/PostgreSQL.

Exactly-once processing of messages. Holds all queues in a single table. Supports psycopg2 driver.

Benchmarks for CockroachDB 3-node cluster on Kubernetes:

• n1-standard-1 nodes with standard disk ~ 300 ops/s
• n1-highcpu-2 nodes with ssd disk ~ 700 ops/s
• n1-highcpu-16 nodes with ssd disk ~ 4500 ops/s

Install

pip3 install pycq

DB Setup

Create the users to use in publishers/consumers:

CREATE USER IF NOT EXISTS <publisher>;
CREATE USER IF NOT EXISTS <consumer>;

Create the queues table:

CREATE TABLE IF NOT EXISTS queue (
  id SERIAL PRIMARY KEY,
  queue STRING NOT NULL,
  data JSONB NOT NULL,
  enqueued_at TIMESTAMP DEFAULT NOW()
);

Create the indexes:

CREATE INDEX ON queue (queue, enqueued_at) STORING (data);

Grant permissions:

GRANT INSERT ON TABLE queue TO <publisher>;
GRANT SELECT,DELETE ON TABLE queue TO <consumer>;

Usage

Publisher

import psycopg2
from pycq import CQ

conn = psycopg2.connect(database=<database>,
                        user=<publisher>,
                        host=<host>,
                        port=26257)

queue = CQ(conn, table="queue")  # Default table is 'queue'
queue.put("example", {"foo": "bar"})

conn.close()

Consumer

import psycopg2
from pycq import CQ

conn = psycopg2.connect(database=<database>,
                        user=<consumer>,
                        host=<host>,
                        port=26257)

queue = CQ(conn)

def handler(message):
    print(f"{message}")
    # {
    #   'data': {'foo': 'bar'},
    #   'enqueued_at': datetime.datetime(2019, 5, 12, 17, 46, 57, 351679)
    # }

try:
    queue.subscribe("example", callback=handler,
        # The following are optional and default values
        # e.g. Poll every 0.5s after a message is found or polling has just started.
        #      Switch to polling every 2s if the queue stays empty for 10s.
        poll_interval=2.0,
        burst_poll_interval=0.5,
        burst_decay_interval=10.0
    )
except KeyboardInterrupt:
    conn.close()