tembo-pgmq-python 0.8.0

Python client for the PGMQ Postgres extension.

Tembo's Python Client for PGMQ

Installation

Install with pip from pypi.org:

pip install tembo-pgmq-python

Dependencies:

Postgres running the Tembo PGMQ extension.

Usage

Start a Postgres Instance with the Tembo extension installed

docker run -d --name postgres -e POSTGRES_PASSWORD=postgres -p 5432:5432 quay.io/tembo/pg16-pgmq:latest

Using Environment Variables

Set environment variables:

export PG_HOST=127.0.0.1
export PG_PORT=5432
export PG_USERNAME=postgres
export PG_PASSWORD=postgres
export PG_DATABASE=test_db

Initialize a connection to Postgres using environment variables:

from tembo_pgmq_python import PGMQueue, Message

queue = PGMQueue()

Initialize a connection to Postgres without environment variables

from tembo_pgmq_python import PGMQueue, Message

queue = PGMQueue(
    host="0.0.0.0",
    port="5432",
    username="postgres",
    password="postgres",
    database="postgres"
)

Create a queue

queue.create_queue("my_queue")

or a partitioned queue

queue.create_partitioned_queue("my_partitioned_queue", partition_interval=10000)

List all queues

queues = queue.list_queues()
for q in queues:
    print(f"Queue name: {q}")

Send a message

msg_id: int = queue.send("my_queue", {"hello": "world"})

Send a batch of messages

msg_ids: list[int] = queue.send_batch("my_queue", [{"hello": "world"}, {"foo": "bar"}])

Read a message, set it invisible for 30 seconds

read_message: Message = queue.read("my_queue", vt=30)
print(read_message)

Read a batch of messages

read_messages: list[Message] = queue.read_batch("my_queue", vt=30, batch_size=5)
for message in read_messages:
    print(message)

Read messages with polling

The read_with_poll method allows you to repeatedly check for messages in the queue until either a message is found or the specified polling duration is exceeded. This can be useful in scenarios where you want to wait for new messages to arrive without continuously querying the queue in a tight loop.

In the following example, the method will check for up to 5 messages in the queue my_queue, making the messages invisible for 30 seconds (vt), and will poll for a maximum of 5 seconds (max_poll_seconds) with intervals of 100 milliseconds (poll_interval_ms) between checks.

read_messages: list[Message] = queue.read_with_poll("my_queue", vt=30, qty=5, max_poll_seconds=5, poll_interval_ms=100)
for message in read_messages:
    print(message)

This method will continue polling until it either finds the specified number of messages (qty) or the max_poll_seconds duration is reached. The poll_interval_ms parameter controls the interval between successive polls, allowing you to avoid hammering the database with continuous queries.

Archive the message after we're done with it. Archived messages are moved to an archive table

archived: bool = queue.archive("my_queue", read_message.msg_id)

Archive a batch of messages

archived_ids: list[int] = queue.archive_batch("my_queue", [msg_id1, msg_id2])

Delete a message completely

read_message: Message = queue.read("my_queue")
deleted: bool = queue.delete("my_queue", read_message.msg_id)

Delete a batch of messages

deleted_ids: list[int] = queue.delete_batch("my_queue", [msg_id1, msg_id2])

Set the visibility timeout (VT) for a specific message

updated_message: Message = queue.set_vt("my_queue", msg_id, 60)
print(updated_message)

Pop a message, deleting it and reading it in one transaction

popped_message: Message = queue.pop("my_queue")
print(popped_message)

Purge all messages from a queue

purged_count: int = queue.purge("my_queue")
print(f"Purged {purged_count} messages from the queue.")

Detach an archive from a queue

queue.detach_archive("my_queue")

Drop a queue

dropped: bool = queue.drop_queue("my_queue")
print(f"Queue dropped: {dropped}")

Validate the length of a queue name

queue.validate_queue_name("my_queue")

Get queue metrics

The metrics method retrieves various statistics for a specific queue, such as the queue length, the age of the newest and oldest messages, the total number of messages, and the time of the metrics scrape.

metrics = queue.metrics("my_queue")
print(f"Metrics: {metrics}")

Access individual metrics

You can access individual metrics directly from the metrics method's return value:

metrics = queue.metrics("my_queue")
print(f"Queue name: {metrics.queue_name}")
print(f"Queue length: {metrics.queue_length}")
print(f"Newest message age (seconds): {metrics.newest_msg_age_sec}")
print(f"Oldest message age (seconds): {metrics.oldest_msg_age_sec}")
print(f"Total messages: {metrics.total_messages}")
print(f"Scrape time: {metrics.scrape_time}")

Get metrics for all queues

The metrics_all method retrieves metrics for all queues, allowing you to iterate through each queue's metrics.

all_metrics = queue.metrics_all()
for metrics in all_metrics:
    print(f"Queue name: {metrics.queue_name}")
    print(f"Queue length: {metrics.queue_length}")
    print(f"Newest message age (seconds): {metrics.newest_msg_age_sec}")
    print(f"Oldest message age (seconds): {metrics.oldest_msg_age_sec}")
    print(f"Total messages: {metrics.total_messages}")
    print(f"Scrape time: {metrics.scrape_time}")