More flexible PGMQ Postgres extension Python client that using sqlalchemy ORM, supporting both async and sync engines, sessionmakers or built from dsn.
- Supports async and sync
enginesandsessionmakers, or built fromdsn. - Automatically creates
pgmq(orpg_partman) extension on the database if not exists. - Supports all postgres DBAPIs supported by sqlalchemy.
e.g.
psycopg,psycopg2,asyncpg..
See SQLAlchemy Postgresql Dialects
Install with pip:
pip install pgmq-sqlalchemyInstall with additional DBAPIs packages:
pip install "pgmq-sqlalchemy[asyncpg]" pip install "pgmq-sqlalchemy[psycopg2-binary]" # pip install "pgmq-sqlalchemy[postgres-python-driver]"Prerequisites: Postgres with PGMQ extension installed.
For quick setup:
docker run -d --name postgres -e POSTGRES_PASSWORD=postgres -p 5432:5432 quay.io/tembo/pg16-pgmq:latestFor more information, see PGMQ
Note
Check pgmq-sqlalchemy Document for more examples and detailed usage.
For dispatcher.py:
from typing import List from pgmq_sqlalchemy import PGMQueue postgres_dsn = 'postgresql://postgres:postgres@localhost:5432/postgres' pgmq = PGMQueue(dsn=postgres_dsn) pgmq.create_queue('my_queue') msg = {'key': 'value', 'key2': 'value2'} msg_id:int = pgmq.send('my_queue', msg) # could also send a list of messages msg_ids:List[int] = pgmq.send_batch('my_queue', [msg, msg])For consumer.py:
from pgmq_sqlalchemy import PGMQueue from pgmq_sqlalchemy.schema import Message postgres_dsn = 'postgresql://postgres:postgres@localhost:5432/postgres' pgmq = PGMQueue(dsn=postgres_dsn) # read a single message msg:Message = pgmq.read('my_queue') # read a batch of messages msgs:List[Message] = pgmq.read_batch('my_queue', 10)For monitor.py:
from pgmq_sqlalchemy import PGMQueue from pgmq_sqlalchemy.schema import QueueMetrics postgres_dsn = 'postgresql://postgres:postgres@localhost:5432/postgres' pgmq = PGMQueue(dsn=postgres_dsn) # get queue metrics metrics:QueueMetrics = pgmq.metrics('my_queue') print(metrics.queue_length) print(metrics.total_messages)Welcome to open an issue or pull request !
See Development on Online Document or CONTRIBUTING.md for more information.
- Add time-based partition option and validation to
create_partitioned_queuemethod. - Read(single/batch) Archive Table (
read_archivemethod ) - Detach Archive Table (
detach_archivemethod ) - Add
set_vtutils method.