pgclient - yet another pool-based python2/3 compatible psycopg2 wrapper.
The client is fully based on thread-safe reliable connections pool and safe transactions executing
Tested on python2.7+, python3.4+
- python-dev
- libpq-dev
pip install pgclient
from pgclient import PostgresClient
pg_client = PostgresClient(dsn='user=admin password=admin dbname=test host=localhost port=5432')
# OR
pg_client = PostgresClient(username='test', password='test', ...)
with self.pg_client.get_cursor() as cursor:
cursor.execute('SELECT * FROM MyTable')
result_set = cursor.fetchall()
Assume that we use the following sql schema:
CREATE TABLE users (
id SERIAL,
username VARCHAR NOT NULL
)
Cursor context manager
with self.pg_client.get_cursor() as cursor:
cursor.execute('SELECT * FROM users')
users = cursor.fetchall()
username = users[0]['username']
NOTE: Default cursor_factory is psycopg2.extras.RealDictCursor
To override default factory, there are two ways:
-
Override default one for client instance
pg_client = PostgresClient(..., cursor_factory=psycopg2.extras.NamedTupleCursor) -
Override for context
with pg_client.get_cursor(cursor_factory=MyCursor) as cursor: cursor.execute('SELECT * FROM users')
All requests inside with context will be executed and automatically committed within one transaction
(or rolled back in case if database errors)
with self.pg_client.get_cursor() as transaction:
transaction.execute('INSERT INTO users VALUES name="Mark"')
transaction.execute('INSERT INTO users VALUES name="Paolo"')
transaction.execute('SELECT * FROM users')
users = transaction.fetchall()
Starting a new transaction, it guarantees that connection is alive
with self.pg_client.get_cursor() as cursor:
# connection is alive
cursor.execute(...)
# Or manually
conn = self.pg_client.acquire_conn()
conn.execute(...)
...
self.pg_client.release_conn(conn)
Instead of basic psycopg2.Error based errors, Extended exception classes have been added.
So now you will get more meaningful error information in case of any errors during
the postgres communication and use error handling in more flexible way.
Example:
from pgclient import exceptions as pg_exc
try:
with self.pg_client.get_cursor() as transaction:
transaction.execute(...)
except pg_exc.IntegrityConstraintViolation as err:
logger.error(err.message, err.diag, err.pgcode)
except pg_exc.DataException as err:
...
To catch all errors:
try:
with self.pg_client.get_cursor() as transaction:
transaction.execute(...)
except pg_exc.PgClientError as err:
logger.error(err)
...
To run integration test you need to install the following:
Run system test:
- Run postgresql container:
docker-compose up -d postgresql - Run system tests:
make system_test - Stop postgresql container:
docker-compose stop postgresql
To test with postgresql:9.0 run postgresql_90 container with docker compose.
Both versions are being tested with travis ci.
Warm welcome to suggestions and concerns