Database Tools

./scripts/psql-connect.sh --cluster <cluster-id> --command "SELECT * FROM entities LIMIT 10"

from safe_query import build, regex, ident, TENANT_SLUG

# Simple SELECT — user-supplied tenant_id goes through a validator
sql = build(
    "SELECT * FROM {tbl} WHERE tenant_id = {tid} LIMIT 10",
    tbl=ident("entities"),
    tid=regex(tenant_id, TENANT_SLUG),
)
# Pass `sql` to your driver: psycopg `cur.execute(sql)`, pgx `conn.Query(ctx, sql)`, etc.

# Aggregate query (no user-supplied values)
sql = build(
    "SELECT tenant_id, COUNT(*) as count FROM objectives GROUP BY tenant_id",
)

# Join query — declare e/o as table aliases after each ident() expansion
sql = build(
    "SELECT e.entity_id, e.name, o.title "
    "FROM {e} e INNER JOIN {o} o ON e.entity_id = o.entity_id "
    "WHERE e.tenant_id = {tid}",
    e=ident("entities"),
    o=ident("objectives"),
    tid=regex(tenant_id, TENANT_SLUG),
)

./scripts/psql-connect.sh --cluster <cluster-id> --admin --command "ALTER TABLE entities ADD COLUMN status VARCHAR(50)"

# Create table with index — TWO transactions, in order
conn.execute("CREATE TABLE IF NOT EXISTS entities (...)")           # tx 1: DDL
conn.execute("CREATE INDEX ASYNC idx_entities_tenant ON entities(tenant_id)")  # tx 2: DDL

# Insert rows — build each statement with safe_query.
from safe_query import build, allow, regex, literal, UUID, TENANT_SLUG

with conn.transaction():
    for row in rows:  # keep each transaction under 3,000 rows
        sql = build(
            "INSERT INTO entities (entity_id, tenant_id, name) "
            "VALUES ({eid}, {tid}, {name})",
            eid=regex(row["entity_id"], UUID),
            tid=regex(row["tenant_id"], TENANT_SLUG),
            name=literal(row["name"]),
        )
        conn.execute(sql)

# Two-step column migration
STATUSES = {"active", "archived", "pending"}
conn.execute("ALTER TABLE entities ADD COLUMN status VARCHAR(50)")  # tx 1
sql = build(
    "UPDATE entities SET status = {s} "
    "WHERE status IS NULL AND tenant_id = {tid}",
    s=allow("active", STATUSES),
    tid=regex(tenant_id, TENANT_SLUG),
)
conn.execute(sql)                                                   # tx 2

SELECT table_name FROM information_schema.tables WHERE table_schema = 'public';

SELECT column_name, data_type, is_nullable, column_default
FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = 'entities'
ORDER BY ordinal_position;

SELECT indexname, indexdef FROM pg_indexes WHERE schemaname = 'public' AND tablename = 'entities';