psql for daily work: meta-commands and scripts

describe_session.py

"""Finding your way around a database with psql's describe commands and expanded output."""

from pg_session import Server

with Server(settings={"autovacuum": "off"}) as pg:
    pg.up()
    pg.shell("createdb", "shop")
    pg.psql("setup.sql", db="shop", flags=("-q",))
    pg.psql("describe.sql", db="shop")

$ createdb shop

$ psql -d shop -X -q -f setup.sql

$ psql -d shop -X -a -f describe.sql
\dt+
                                         List of tables
 Schema |   Name    | Type  |  Owner   | Persistence | Access method |    Size    | Description 
--------+-----------+-------+----------+-------------+---------------+------------+-------------
 public | audit_log | table | postgres | permanent   | heap          | 8192 bytes | 
 public | customers | table | postgres | permanent   | heap          | 16 kB      | 
 public | orders    | table | postgres | permanent   | heap          | 184 kB     | 
(3 rows)

\di+
                                                   List of indexes
 Schema |        Name         | Type  |  Owner   |   Table   | Persistence | Access method |    Size    | Description 
--------+---------------------+-------+----------+-----------+-------------+---------------+------------+-------------
 public | audit_log_pkey      | index | postgres | audit_log | permanent   | btree         | 8192 bytes | 
 public | customers_pkey      | index | postgres | customers | permanent   | btree         | 16 kB      | 
 public | orders_customer_idx | index | postgres | orders    | permanent   | btree         | 40 kB      | 
 public | orders_pkey         | index | postgres | orders    | permanent   | btree         | 88 kB      | 
(4 rows)

\d orders
                               Table "public.orders"
   Column    |     Type      | Collation | Nullable |           Default            
-------------+---------------+-----------+----------+------------------------------
 id          | bigint        |           | not null | generated always as identity
 customer_id | integer       |           | not null | 
 total       | numeric(10,2) |           | not null | 
 placed      | date          |           | not null | 
Indexes:
    "orders_pkey" PRIMARY KEY, btree (id)
    "orders_customer_idx" btree (customer_id)
Foreign-key constraints:
    "orders_customer_id_fkey" FOREIGN KEY (customer_id) REFERENCES customers(id)

-- \x auto turns a result wider than the screen into one block per row. psql writing to a file has no
-- screen width, so \pset columns sets one: 80 characters.
\pset columns 80
Target width is 80.
\x auto
Expanded display is used automatically.
SELECT name, setting, unit, context, short_desc FROM pg_settings WHERE name IN ('work_mem', 'lock_timeout');
-[ RECORD 1 ]---------------------------------------------------------
name       | lock_timeout
setting    | 0
unit       | ms
context    | user
short_desc | Sets the maximum allowed duration of any wait for a lock.
-[ RECORD 2 ]---------------------------------------------------------
name       | work_mem
setting    | 4096
unit       | kB
context    | user
short_desc | Sets the maximum memory to be used for query workspaces.

SELECT name, setting, unit FROM pg_settings WHERE name IN ('work_mem', 'lock_timeout');
     name     | setting | unit 
--------------+---------+------
 lock_timeout | 0       | ms
 work_mem     | 4096    | kB
(2 rows)

\x off
Expanded display is off.

setup.sql

-- The shop database of this lesson: three customers, 3,000 orders and an empty audit log.
CREATE TABLE customers (id int PRIMARY KEY, name text NOT NULL, city text NOT NULL);
CREATE TABLE orders (
  id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_id int NOT NULL REFERENCES customers,
  total numeric(10,2) NOT NULL,
  placed date NOT NULL
);
CREATE INDEX orders_customer_idx ON orders (customer_id);
CREATE TABLE audit_log (id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY, note text NOT NULL);
INSERT INTO customers VALUES (1, 'Asha Rao', 'Pune'), (2, 'Imran Khan', 'Lucknow'), (3, 'Meera Iyer', 'Chennai');
INSERT INTO orders (customer_id, total, placed)
SELECT 1 + n % 3, 100 + n % 900, date '2026-01-01' + n % 200 FROM generate_series(1, 3000) AS n;
ANALYZE orders;
CREATE ROLE clerk LOGIN;
GRANT SELECT, INSERT ON customers, orders, audit_log TO clerk;

describe.sql

\dt+
\di+
\d orders
-- \x auto turns a result wider than the screen into one block per row. psql writing to a file has no
-- screen width, so \pset columns sets one: 80 characters.
\pset columns 80
\x auto
SELECT name, setting, unit, context, short_desc FROM pg_settings WHERE name IN ('work_mem', 'lock_timeout');
SELECT name, setting, unit FROM pg_settings WHERE name IN ('work_mem', 'lock_timeout');
\x off

pg_session.py

"""Throwaway PostgreSQL 18.6 servers for the examples of this lesson.

Every session creates a cluster of its own with initdb in a new folder inside the examples folder, starts it, runs
psql scripts and server programs on it, prints what they printed, and stops and deletes it again, also when something
fails. The server is the pinned PostgreSQL 18.6 from the mise installations next to the Python that runs this file.
It listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system reports as
unused, with no Unix socket, and it asks for a password (scram-sha-256) made up for the run and never printed.
learn:verify's run sandbox cannot read the home folder, where mise keeps PostgreSQL, so these sessions are recorded
with LEARN_VERIFY_NETWORK=allow; they read only this folder and the toolchain and connect to nothing but 127.0.0.1.

The output is exactly what the programs printed, with four substitutions so that a record never shows the computer
it ran on and is the same on every run: the run's folder is shown as ".", the folder of PostgreSQL's programs is left
out of their paths, the operating-system user's name is shown as "learner", and the dots pg_ctl prints while it waits
(one about every tenth of a second) as three. The sessions never print the port. A session stops with status 1 when
a command fails that should have worked, or reports a different number of errors than the session expects, so a run
that went wrong is never recorded as if it had worked.
"""

import os
import pwd
import re
import secrets
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import textwrap
import time
from pathlib import Path

PG_VERSION = "18.6"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
BIN = INSTALLS / "postgres" / PG_VERSION / "bin"
OS_USER = pwd.getpwuid(os.geteuid()).pw_name


def free_port():
    """A TCP port on 127.0.0.1 that nothing listens on now."""
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


class Server:
    """One throwaway cluster: initdb, settings, start; stopped and deleted when the with block ends."""

    def __init__(self, settings=None, initdb=(), name="data"):
        self.settings = dict(settings or {})
        self.initdb_args = list(initdb)
        self.name = name
        self.running = []

    def __enter__(self):
        if not (BIN / "postgres").exists():
            sys.exit(f"PostgreSQL {PG_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="pg", dir=".")).resolve()
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        (self.base / "pwfile").write_text(self.password + "\n")
        self.env = {**os.environ, "PGHOST": "127.0.0.1", "PGPORT": str(self.port), "PGUSER": "postgres",
                    "PGPASSWORD": self.password, "PGDATABASE": "postgres", "PGCONNECT_TIMEOUT": "10"}
        for key in ("PGSERVICE", "PGSERVICEFILE", "PGOPTIONS", "PSQLRC"):
            self.env.pop(key, None)
        for script in Path(".").glob("*.sql"):
            shutil.copy(script, self.base / script.name)
        for data in Path(".").glob("*.csv"):
            shutil.copy(data, self.base / data.name)
        return self

    def __exit__(self, *exc):
        for folder, _port in reversed(self.running):
            self._call(["pg_ctl", "-D", folder, "-m", "fast", "-w", "-t", "60", "stop"])
        self.running = []
        shutil.rmtree(self.base, ignore_errors=True)

    # ---- the cluster ---------------------------------------------------------------------------------------------
    def initdb(self, show=False):
        args = ["initdb", "-D", self.name, "-U", "postgres", "--auth=scram-sha-256", "--pwfile=pwfile",
                *self.initdb_args]
        if not show:
            args.append("--no-instructions")
        self._expect(args, show=show)
        with open(self.base / self.name / "postgresql.conf", "a") as conf:
            conf.write("\n# Added by the session: 127.0.0.1 only, no Unix socket.\n")
            conf.write(f"listen_addresses = '127.0.0.1'\nport = {self.port}\nunix_socket_directories = ''\n")
            for key, value in self.settings.items():
                conf.write(f"{key} = {value}\n")

    def start(self, show=False, folder=None, port=None):
        folder = folder or self.name
        self._expect(["pg_ctl", "-D", folder, "-l", f"{folder}.log", "-w", "-t", "60", "start"], show=show)
        self.running.append((folder, port or self.port))

    def up(self):
        """initdb and start, quietly."""
        self.initdb()
        self.start()
        return self

    def standby(self, folder="standby"):
        """A streaming standby of this server, made with pg_basebackup -R, started on a port of its own."""
        self._expect(["pg_basebackup", "-D", folder, "-R", "-X", "stream", "-c", "fast"])
        port = free_port()
        with open(self.base / folder / "postgresql.conf", "a") as conf:
            conf.write(f"port = {port}\n")
        self.start(folder=folder, port=port)
        for _ in range(300):
            if self.query("SELECT count(*) FROM pg_stat_replication WHERE state = 'streaming'") == "1":
                return port
            time.sleep(0.1)
        sys.exit("the standby did not start streaming")

    # ---- running things ------------------------------------------------------------------------------------------
    def _call(self, args, env=None):
        args = [str(a) for a in args]
        done = subprocess.run([str(BIN / args[0]), *args[1:]], stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                              text=True, cwd=self.base, env=env or self.env)
        return done.returncode, self.clean(done.stdout)

    def _expect(self, args, show=False, status=0):
        code, out = self._call(args)
        if show:
            _show(args, out)
        if code != status:
            if not show:
                print(out)
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """The run's folder as ".", the user as learner, pg_ctl's waiting dots as three (see the docstring)."""
        text = text.replace(str(BIN) + "/", "").replace(str(self.base) + "/", "").replace(str(self.base), ".")
        text = re.sub(r"(waiting for server to (?:start|shut down))\.+", r"\1...", text)
        return text.replace(f'"{OS_USER}"', '"learner"')

    def wait_for(self, sql, value):
        """Waits (unprinted) until a query returns the value, for example after a reload."""
        for _ in range(300):
            if self.query(sql) == value:
                return
            time.sleep(0.1)
        sys.exit(f"{sql} never returned {value}")

    def show_file(self, path):
        """Prints a text file of the cluster, as `cat` would."""
        print(f"$ cat {path}")
        print((self.base / path).read_text().rstrip("\n"))
        print()

    def query(self, sql, port=None):
        """One value from one query, unprinted (the session uses it to wait or to decide)."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        code, out = self._call(["psql", "-X", "-At", "-c", sql], env=env)
        if code != 0:
            sys.exit(f"the query failed: {out}")
        return out.strip()

    def silent(self, sql):
        """Runs a statement without printing it (used only to give a role the run's password)."""
        self.query(sql)

    def psql(self, script, errors=0, status=None, user=None, db=None, port=None, flags=("-a",), note=None):
        """Runs a psql script with -X (no ~/.psqlrc) and the given flags, and prints the command and its output.
        `errors` is how many ERROR lines the script must print; `status` the exit status it must end with."""
        env = {**self.env, "PGPORT": str(port or self.port)}
        if user:
            env["PGUSER"] = user
        if db:
            env["PGDATABASE"] = db
        args = ["psql", "-X", *flags, "-f", script]
        code, out = self._call(args, env=env)
        shown = ["psql", "-X", *flags, "-f", script]
        if user:
            shown[1:1] = ["-U", user]
        if db:
            shown[1:1] = ["-d", db]
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if "ERROR:" in line or ": error:" in line)
        expected_status = 0 if status is None else status
        if found != errors or code != expected_status:
            sys.exit(f"{script}: {found} errors and status {code}; the session expects {errors} and {expected_status}")

    def sessions(self, count):
        """Opens `count` idle psql connections (each one a client backend) until the with block ends."""
        return _Idle(self, count)

    def shell(self, *args, status=0):
        """Runs one of PostgreSQL's programs and prints the command and its output."""
        return self._expect(list(args), show=True, status=status)

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the cluster (sorted, folders with a slash), as `ls -p` would list them."""
        path = self.base / folder
        names = sorted(n + ("/" if (path / n).is_dir() else "") for n in os.listdir(path) if n.startswith(prefix))
        print(f"-- in {folder}" + (f", the names that start with {prefix}" if prefix else "") + ":")
        print(textwrap.fill("  ".join(names), width=78, initial_indent="   ", subsequent_indent="   ",
                            break_on_hyphens=False))
        print()
        return names


def _arg(a):
    """A command-line argument as a reader would type it: double quotes around text with single quotes in it."""
    a = str(a)
    if re.fullmatch(r"[\w@%+=:,./-]+", a):
        return a
    if not re.search(r"[\"$`\\!]", a):
        return f'"{a}"'
    return shlex.quote(a)


def _show(args, out):
    print("$ " + " ".join(_arg(a) for a in args))
    if out.strip():
        print(out.rstrip("\n"))
    print()


class _Idle:
    """Idle psql connections, closed again when the with block ends."""

    def __init__(self, server, count):
        self.server, self.count, self.procs = server, count, []

    def __enter__(self):
        before = int(self.server.query("SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend'"))
        for _ in range(self.count):
            self.procs.append(subprocess.Popen([str(BIN / "psql"), "-X", "-q"], stdin=subprocess.PIPE,
                                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                                               env=self.server.env, cwd=self.server.base))
        sql = "SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'client backend'"
        for _ in range(300):
            # the query's own connection is one more client backend
            if int(self.server.query(sql)) == before + self.count:
                return self
            time.sleep(0.1)
        sys.exit("the idle connections did not open")

    def __exit__(self, *exc):
        for p in self.procs:
            p.stdin.close()
            p.wait(timeout=30)

scripts_session.py

"""psql scripts with variables, \\if, \\gexec and ON_ERROR_STOP, run with -a so every line is echoed."""

from pg_session import Server

with Server(settings={"autovacuum": "off"}) as pg:
    pg.up()
    pg.shell("createdb", "shop")
    pg.psql("setup.sql", db="shop", flags=("-q",))
    pg.psql("if_mistake.sql", db="shop", errors=1)
    pg.psql("report.sql", db="shop", errors=1, status=3)
    pg.psql("gexec.sql", db="shop")

$ createdb shop

$ psql -d shop -X -q -f setup.sql

$ psql -d shop -X -a -f if_mistake.sql
-- \if takes a boolean, not an expression: psql does no arithmetic and no comparison.
SELECT count(*) AS n FROM orders \gset
\echo :n orders
3000 orders
\if :n > 1000
psql:if_mistake.sql:4: error: unrecognized value "3000 > 1000" for "\if expression": Boolean expected
  \echo a busy shop
\endif

$ psql -d shop -X -a -f report.sql
-- A report script: stop at the first error, let the server compare, quote every value it splices in.
\set ON_ERROR_STOP on
SELECT count(*) AS n, count(*) > 1000 AS busy FROM orders \gset
\if :busy
  \echo :n orders: a busy shop
3000 orders: a busy shop
\else
  \echo :n orders: a quiet shop
\endif
\set city 'Pune'
SELECT c.name, count(*) AS orders, sum(o.total) AS spent
FROM customers AS c JOIN orders AS o ON o.customer_id = c.id
WHERE c.city = :'city'
GROUP BY c.name;
   name   | orders |   spent   
----------+--------+-----------
 Asha Rao |   1000 | 518800.00
(1 row)

SELECT * FROM order_totals;
psql:report.sql:14: ERROR:  relation "order_totals" does not exist
LINE 1: SELECT * FROM order_totals;
                      ^

$ psql -d shop -X -a -f gexec.sql
-- Turn a catalog query into a maintenance script: one ANALYZE for every table never analysed.
SELECT format('ANALYZE %I.%I', schemaname, relname)
FROM pg_stat_user_tables
WHERE last_analyze IS NULL AND last_autoanalyze IS NULL
ORDER BY relname \gexec
ANALYZE public.audit_log
ANALYZE
ANALYZE public.customers
ANALYZE
SELECT relname, last_analyze IS NOT NULL AS analysed FROM pg_stat_user_tables ORDER BY relname;
  relname  | analysed 
-----------+----------
 audit_log | t
 customers | t
 orders    | t
(3 rows)

if_mistake.sql

-- \if takes a boolean, not an expression: psql does no arithmetic and no comparison.
SELECT count(*) AS n FROM orders \gset
\echo :n orders
\if :n > 1000
  \echo a busy shop
\endif

report.sql

-- A report script: stop at the first error, let the server compare, quote every value it splices in.
\set ON_ERROR_STOP on
SELECT count(*) AS n, count(*) > 1000 AS busy FROM orders \gset
\if :busy
  \echo :n orders: a busy shop
\else
  \echo :n orders: a quiet shop
\endif
\set city 'Pune'
SELECT c.name, count(*) AS orders, sum(o.total) AS spent
FROM customers AS c JOIN orders AS o ON o.customer_id = c.id
WHERE c.city = :'city'
GROUP BY c.name;
SELECT * FROM order_totals;
\echo this line never runs

gexec.sql

-- Turn a catalog query into a maintenance script: one ANALYZE for every table never analysed.
SELECT format('ANALYZE %I.%I', schemaname, relname)
FROM pg_stat_user_tables
WHERE last_analyze IS NULL AND last_autoanalyze IS NULL
ORDER BY relname \gexec
SELECT relname, last_analyze IS NOT NULL AS analysed FROM pg_stat_user_tables ORDER BY relname;

copy_session.py

"""\\copy against server-side COPY, run by the clerk role, which may read and insert rows but not server files."""

from pg_session import Server

with Server(settings={"autovacuum": "off"}) as pg:
    pg.up()
    pg.shell("createdb", "shop")
    pg.psql("setup.sql", db="shop", flags=("-q",))
    pg.silent(f"ALTER ROLE clerk PASSWORD '{pg.password}'")
    pg.psql("copy.sql", db="shop", user="clerk", errors=2)
    pg.show_file("customers_out.csv")

$ createdb shop

$ psql -d shop -X -q -f setup.sql

$ psql -d shop -U clerk -X -a -f copy.sql
-- \copy: psql reads and writes the files, on the computer where psql runs.
\copy customers FROM 'new_customers.csv' WITH (FORMAT csv, HEADER)
COPY 2
\copy (SELECT id, name, city FROM customers ORDER BY id) TO 'customers_out.csv' WITH (FORMAT csv, HEADER)
COPY 5
-- COPY with a file name: the server opens the file, so the role needs the right to read server files.
COPY customers FROM 'new_customers.csv' WITH (FORMAT csv, HEADER);
psql:copy.sql:5: ERROR:  permission denied to COPY from a file
DETAIL:  Only roles with privileges of the "pg_read_server_files" role may COPY from a file.
HINT:  Anyone can COPY to stdout or from stdin. psql's \copy command also works for anyone.
-- A COPY FROM STDIN that fails at once: psql 18.6 skips its in-line data up to the \. line.
COPY customer FROM STDIN WITH (FORMAT csv);
psql:copy.sql:7: ERROR:  relation "customer" does not exist
SELECT count(*) AS customers FROM customers;
 customers 
-----------
         5
(1 row)

$ cat customers_out.csv
id,name,city
1,Asha Rao,Pune
2,Imran Khan,Lucknow
3,Meera Iyer,Chennai
4,Kabir Das,Varanasi
5,Leela Nair,Kochi

copy.sql

-- \copy: psql reads and writes the files, on the computer where psql runs.
\copy customers FROM 'new_customers.csv' WITH (FORMAT csv, HEADER)
\copy (SELECT id, name, city FROM customers ORDER BY id) TO 'customers_out.csv' WITH (FORMAT csv, HEADER)
-- COPY with a file name: the server opens the file, so the role needs the right to read server files.
COPY customers FROM 'new_customers.csv' WITH (FORMAT csv, HEADER);
-- A COPY FROM STDIN that fails at once: psql 18.6 skips its in-line data up to the \. line.
COPY customer FROM STDIN WITH (FORMAT csv);
6,Ravi Kumar,Indore
\.
SELECT count(*) AS customers FROM customers;

new_customers.csv

id,name,city
4,Kabir Das,Varanasi
5,Leela Nair,Kochi

pipeline_session.py

"""psql 18's commands for named prepared statements and pipelines."""

from pg_session import Server

with Server(settings={"autovacuum": "off"}) as pg:
    pg.up()
    pg.shell("createdb", "shop")
    pg.psql("setup.sql", db="shop", flags=("-q",))
    pg.psql("pipeline.sql", db="shop", errors=1)

$ createdb shop

$ psql -d shop -X -q -f setup.sql

$ psql -d shop -X -a -f pipeline.sql
-- A named prepared statement: parse once, bind and run it twice, close it.
SELECT count(*) AS orders, sum(total) AS spent FROM orders WHERE customer_id = $1 \parse orders_of

\bind_named orders_of 1 \g
 orders |   spent   
--------+-----------
   1000 | 518800.00
(1 row)

\bind_named orders_of 2 \g
 orders |   spent   
--------+-----------
   1000 | 519500.00
(1 row)

SELECT name, statement FROM pg_prepared_statements;
   name    |                                     statement                                      
-----------+------------------------------------------------------------------------------------
 orders_of | SELECT count(*) AS orders, sum(total) AS spent FROM orders WHERE customer_id = $1 
(1 row)

\close_prepared orders_of

SELECT count(*) AS prepared FROM pg_prepared_statements;
 prepared 
----------
        0
(1 row)

-- A pipeline: statements go out without waiting for answers; \syncpipeline ends one implicit transaction.
\startpipeline
INSERT INTO audit_log (note) VALUES ('one');
INSERT INTO audit_log (note) VALUES ('two');
\syncpipeline
INSERT INTO audit_log (note) VALUES ('three') RETURNING id;
\endpipeline
INSERT 0 1
INSERT 0 1
 id 
----
  3
(1 row)

INSERT 0 1
-- An error inside a pipeline rolls back its part of the pipeline and skips the rest of it.
\startpipeline
INSERT INTO audit_log (note) VALUES ('four');
SELECT 1 / 0;
INSERT INTO audit_log (note) VALUES ('five');
\endpipeline
INSERT 0 1
psql:pipeline.sql:20: ERROR:  division by zero
SELECT id, note FROM audit_log ORDER BY id;
 id | note  
----+-------
  1 | one
  2 | two
  3 | three
(3 rows)

pipeline.sql

-- A named prepared statement: parse once, bind and run it twice, close it.
SELECT count(*) AS orders, sum(total) AS spent FROM orders WHERE customer_id = $1 \parse orders_of
\bind_named orders_of 1 \g
\bind_named orders_of 2 \g
SELECT name, statement FROM pg_prepared_statements;
\close_prepared orders_of
SELECT count(*) AS prepared FROM pg_prepared_statements;
-- A pipeline: statements go out without waiting for answers; \syncpipeline ends one implicit transaction.
\startpipeline
INSERT INTO audit_log (note) VALUES ('one');
INSERT INTO audit_log (note) VALUES ('two');
\syncpipeline
INSERT INTO audit_log (note) VALUES ('three') RETURNING id;
\endpipeline
-- An error inside a pipeline rolls back its part of the pipeline and skips the rest of it.
\startpipeline
INSERT INTO audit_log (note) VALUES ('four');
SELECT 1 / 0;
INSERT INTO audit_log (note) VALUES ('five');
\endpipeline
SELECT id, note FROM audit_log ORDER BY id;