connect() built the pool with `await asyncpg.create_pool(...)` outside the try that wraps SELECT-1 validation - with the default min_size=1 the pool connects eagerly, so a bad host/auth propagated with no log.exception line, breaking the documented "every method logs then re-raises" contract. create_pool() now sits inside the same try/log/re-raise as validation, and either failure point tears down a partially-built pool before re-raising. close() didn't take _connect_lock while connect() did, so a close() racing an in-flight connect() would see _pool is None and no-op as success while connect() went on to install a live pool - a shutdown handler racing a reconnect could "close" the instance while real connections stayed open. close() now takes the same lock via a shared _close_locked() helper. Signed-off-by: disqualifier <dev@disqualifier.me>
125 lines
5.1 KiB
Markdown
125 lines
5.1 KiB
Markdown
# psql
|
|
|
|
Async PostgreSQL wrapper over [asyncpg](https://magicstack.github.io/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.7
|
|
```
|
|
|
|
Direct:
|
|
|
|
```bash
|
|
pip install "psql @ git+ssh://git@git.rethinkstudios.io/rethink-public/psql.git@v0.1.7"
|
|
```
|
|
|
|
Pulls `asyncpg`.
|
|
|
|
Drop the `@v0.1.7` 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
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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
|
|
|
|
```python
|
|
# 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.
|