Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -17,3 +17,6 @@ docs/superpowers/

# Local follow-up tracking (not shipped with the package)
/FOLLOWUP.md

# Git worktrees
/.worktrees/
41 changes: 41 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -48,3 +48,44 @@ podman/docker, pgdevkit first checks (with a short timeout) whether Postgres
is already reachable at that address and skips container management if so.
Set `PGDEVKIT_SKIP_CONTAINER=1` to always assume it's already there and skip
that check too.

## `pgdevkit.db` — helpers for application code

Install with the `db` extra: `pip install pgdevkit[db]`.

- **`PostgresTableModel`** — a `pydantic.BaseModel` base class for models
that map 1:1 to a table row. Implement `get_table_name()` (returns
`(schema, table)`) and `get_primary_key()` on each model.
- **`PgPool`** — an async connection pool keyed off
`{env_prefix}HOST/PORT/DB/USER/PASSWORD` env vars. Call `await pool.open()`
once at startup, then use `async with pool.connection() as con:`.
- **CRUD functions** — `pg_retrieve`, `pg_retrieve_many`, `pg_insert`,
`pg_insert_many`, `pg_update`, `pg_update_dict`, `pg_upsert`,
`pg_upsert_dict`, `pg_upsert_many`, `pg_upsert_many_dict`, `pg_delete`,
`pg_delete_dict` — typed (`PostgresTableModel`-based) or dict-based CRUD
against a table, built on `psycopg` for safe identifier/value handling.
- **`SqlLoader`** — loads and caches `.sql` files from
`{root}/<topic>/<name>.sql`, for keeping hand-written queries out of
Python source.

```python
from pgdevkit.db import PgPool, PostgresTableModel, pg_retrieve, pg_upsert

class Widget(PostgresTableModel):
id: int
name: str

@staticmethod
def get_table_name() -> tuple[str, str]:
return ("public", "widget")

@staticmethod
def get_primary_key() -> list[str]:
return ["id"]

pool = PgPool(env_prefix="POSTGRES_")
await pool.open()
async with pool.connection() as con:
widget = await pg_retrieve(con, Widget, {"id": 1})
await pg_upsert(con, Widget(id=1, name="thing"), Widget)
```
Loading