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.5

Direct:

pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.5"

Pulls asyncpg.

Drop the @v0.1.5 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 asyncpg Record objects — identical shape to the mysql lib.
• Values are always parameterized. Layer 1 builds $1, $2 internally; Layer 2 takes your $1 placeholders + 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: pool property → 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.