AWS CloudShell内のpsqlとPython(Psycopg)からAurora DSQLに接続してみた

$ aws --version
aws-cli/2.22.12 Python/3.12.6 Linux/6.1.115-126.197.amzn2023.x86_64 exec-env/CloudShell exe/x86_64.amzn.2023

$ psql --version
psql (PostgreSQL) 15.9

$ export HOST=xxx.dsql.us-east-1.on.aws
$ aws dsql generate-db-connect-admin-auth-token \
  --region us-east-1 \
  --expires-in 3600 \
  --hostname $HOST

caabtv4vbsxldsnxd3nzelop4a.dsql.us-east-1.on.aws/?Action=DbConnectAdmin&...&X-Amz-Signature=xxx

$ export PGPASSWORD=$(aws dsql generate-db-connect-admin-auth-token \
--region us-east-1 \
--expires-in 3600 \
--hostname $HOST)

$ export PGSSLMODE=require

$ psql \
--username admin \
--dbname postgres \
--host $HOST

psql (15.9, server 16.5)
WARNING: psql major version 15, server major version 16.
         Some psql features might not work.
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_128_GCM_SHA256, compression: off)
Type "help" for help.

postgres=> 

$ curl -LsSf https://astral.sh/uv/install.sh | sh
downloading uv 0.5.9 x86_64-unknown-linux-gnu
no checksums to verify
installing to /home/cloudshell-user/.local/bin
  uv
  uvx
everything's installed!

$ uv init dsql -p 3.13
Initialized project `dsql` at `/home/cloudshell-user/dsql`

$ cd dsql

$ uv add boto3 "psycopg[binary]>=3"

import psycopg
import boto3

def main(cluster_endpoint):
    region = 'us-east-1' # XXX

    # Generate a password token
    client = boto3.client("dsql", region_name=region)
    password_token = client.generate_db_connect_admin_auth_token(cluster_endpoint, region)

    conn = psycopg.connect(
        dbname='postgres',
        user='admin',
        host=cluster_endpoint,
        password=password_token,
        sslmode='require',
        autocommit=True,
    )

    cur = conn.cursor()

    cur.execute(b"""
        CREATE TABLE IF NOT EXISTS owner(
            id uuid NOT NULL DEFAULT gen_random_uuid(),
            name varchar(30) NOT NULL,
            city varchar(80) NOT NULL,
            telephone varchar(20) DEFAULT NULL,
            PRIMARY KEY (id))"""
        )

    # Insert some rows
    cur.execute("INSERT INTO owner(name, city, telephone) VALUES('John Doe', 'Anytown', '555-555-1999')")

    cur.execute("SELECT * FROM owner WHERE name='John Doe'")
    row = cur.fetchone()

    print(row)

    # Placing this cleanup the table after the example. If we run the example
    # again we do not have to worry about data inserted by previous runs
    cur.execute("DELETE FROM owner where name = 'John Doe'")

if __name__ == "__main__":
    # Replace with your own cluster's endpoint
    cluster_endpoint = "xxx.dsql.us-east-1.on.aws" # XXX
    main(cluster_endpoint)

$ uv run test.py
(UUID('c26dd462-23d9-4241-9812-fcace766d5f3'), 'John Doe', 'Anytown', '555-555-1999')