Python client for the PGMQ Postgres extension.
Project description
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}")
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
Built Distribution
Hashes for tembo_pgmq_python-0.8.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 72905b6d421f4a4394e97e0f116ec2fe5e2fc1c8c2f433ab5d5aa72c4a7c4cad |
|
MD5 | 3ce1753f82c20de3ce6119abc83342db |
|
BLAKE2b-256 | becb83200a7facbaaa9cc2d1dbc0228a63fd6f98b2c8aa326c80fbcf2b974efa |