PostgreSQL wire: Python (psycopg 3)

The PostgreSQL wire is how most BI tools reach MachWarehouse. This program is ordinary psycopg code: CREATE TABLE, COPY FROM STDIN, an aggregate.

The only change

# host
 postgres.internal
 warehouse.internal
# port
 5432
 5432 (PostgreSQL wire)
# user / password / database
 PostgreSQL values
 MachWarehouse values

DATABASE_URL: before postgresql://app:...@postgres.internal:5432/demo?sslmode=verify-full, after postgresql://app:...@warehouse.internal:5432/demo?sslmode=verify-full.

Run

pip install -r requirements.txt
DATABASE_URL=postgresql://app:...@warehouse.internal:5432/demo python main.py

Library: psycopg 3.3.6.

Source

The program as it is kept and tested in the Mach Database Warehouse repository; highlighted lines read the connection setting.

main.py

"""Load the demo events with COPY and aggregate over the PostgreSQL wire (psycopg 3). Plain psycopg code."""

import os
import pathlib

import psycopg

DATASET = pathlib.Path(__file__).resolve().parents[2] / "dataset"


def main() -> None:
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        conn.execute((DATASET / "postgres.sql").read_text())
        with conn.cursor() as cur:
            with cur.copy("COPY events FROM STDIN (FORMAT csv, HEADER true)") as copy:
                copy.write((DATASET / "events.csv").read_bytes())
            cur.execute(
                "SELECT country, count(*) AS events, sum(revenue) AS revenue "
                "FROM events GROUP BY country ORDER BY revenue DESC"
            )
            for country, events, revenue in cur:
                print(f"{country}\t{events}\t{revenue}")


if __name__ == "__main__":
    main()

requirements.txt

psycopg[binary]==3.3.6