5.1 KiB
psql
Async PostgreSQL wrapper over asyncpg — a small,
config-free, fail-loud, two-layer API: friendly portable verbs for the common case, a
raw escape hatch for the rest. Second of the datastore trio (redis / psql / mysql),
a sibling of the mongo lib. Class is PsqlDB.
Install
requirements.txt:
psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.6
Direct:
pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.6"
Pulls asyncpg.
Drop the @v0.1.6 suffix from the line above to install the latest unpinned.
The two-layer API
Layer 1 — friendly verbs for simple single-table CRUD. These hide the dialect and are
identical to the mysql lib, so you swap psql↔mysql with zero call-site changes.
Layer 2 — raw escape hatch for the complex ~20% (joins, aggregates, CTEs, window
functions). You write the SQL with $1, $2 placeholders; the wrapper still gives pooling,
parameterization, fail-loud errors, and row→dict conversion. Raw SQL is psql-specific.
There is deliberately nothing in between — no query builder, no ORM. A join goes
through raw fetch(), never a chainable .where()/.join(). (That's SQLAlchemy's job.)
Usage
from psql import PsqlDB
# construction is sync; connect() builds the pool + SELECT 1 to fail loud on bad config
db = await PsqlDB(host="localhost", port=5432, database="app",
user="postgres", password="secret").connect()
# --- layer 1: friendly, portable ---
await db.create_table("users", {"id": "serial primary key", "name": "text not null", "age": "int"})
await db.insert("users", {"name": "ada", "age": 30}) # -> 1 (rowcount)
rows = await db.get("users", {"age": 30}) # [{"id": 1, "name": "ada", ...}]
one = await db.get_one("users", {"name": "ada"}) # {...} or None
ok = await db.exists("users", {"name": "ada"}) # True
await db.delete("users", {"name": "ada"}) # -> rowcount
await db.upsert("users", {"id": 1, "name": "ada2"}, conflict=["id"]) # ON CONFLICT DO UPDATE
await db.close()
Context-manager form:
async with PsqlDB(database="app", user="postgres") as db:
await db.insert("events", {"kind": "login"})
host/port default to None, not "localhost"/5432 — asyncpg only reads a dsn's
embedded host/port when the host/port kwargs are falsy, so passing dsn=... in
pool_kwargs (with no host/port of your own) lets the dsn's server reach asyncpg
instead of being silently overridden. The no-dsn path above still defaults to
localhost:5432 when you don't pass host.
Layer 2 — raw SQL for the complex queries
# a join — not modelled by layer 1, so write it directly. $1 placeholders, plain-dict rows.
rows = await db.fetch(
"SELECT u.name, count(o.id) AS orders "
"FROM users u JOIN orders o ON o.user_id = u.id "
"WHERE u.active = $1 GROUP BY u.name",
True,
)
row = await db.fetchone("SELECT * FROM users WHERE id = $1", 7) # dict or None
n = await db.fetchval("SELECT count(*) FROM users") # scalar or None
await db.execute("UPDATE users SET active = $1 WHERE last_seen < $2", False, cutoff)
# atomic multi-statement block — commits on clean exit, rolls back + re-raises on error
async with db.transaction() as conn:
await conn.execute("INSERT INTO ledger(acct, delta) VALUES($1, $2)", a, -amount)
await conn.execute("INSERT INTO ledger(acct, delta) VALUES($1, $2)", b, +amount)
For anything even Layer 2 doesn't model (COPY, LISTEN/NOTIFY, prepared statements,
cursors), use the raw db.pool — the underlying asyncpg.Pool.
Rows & placeholders
- Rows are plain dicts (
{column: value}), not asyncpgRecordobjects — identical shape to themysqllib. - Values are always parameterized. Layer 1 builds
$1, $2internally; Layer 2 takes your$1placeholders + params. Never f-string a value into SQL. Only identifiers (table/column names) are interpolated, and they're quoted.
Error contract — fail loud
Unlike the mongo lib (which log-and-swallows), this lib re-raises. Every method
catches the driver error (asyncpg.PostgresError / InterfaceError, OSError on
connection loss), logs it via logging.getLogger(__name__), and raises. A None / []
return is only ever a real result (no row, empty table) — never a swallowed failure.
Surface
- Layer 1:
create_database,create_table(name, {col: "type"}),drop(name, table=),insert(table, {col: val}),get(table, {conds}),get_one(table, {conds}),delete(table, {conds}),exists(table, {conds}),upsert(table, {col: val}, conflict=[...]) - Layer 2:
execute(sql, *params),fetch,fetchone,fetchval,transaction() - Raw:
poolproperty →asyncpg.Pool
get/delete/exists/get_one take equality conditions only (col = val AND ...);
anything richer goes through Layer 2.
Versioning
Releases are tagged vX.Y.Z. The install line above pins a release; drop the @vX.Y.Z
suffix to install the latest unpinned. Pin deliberately for reproducible installs.