Using Psycopg3 to interact with Aurora DSQL

このページはお客様の言語に翻訳されていません。 翻訳のリクエスト

This section describes how to use Psycopg3 to interact with Aurora DSQL.

Before you begin, make sure that you have completed the following prerequisites.

• Created a cluster in Aurora DSQL.

• Installed Python. You must be running version 3.8 or higher.

• Created an AWS account and configured the credentials and AWS Region.

• Installed the AWS SDK for Python (Boto3).

Before you get started, install the required dependency.

pip install "psycopg[binary]>=3"

Connect to an Aurora DSQL cluster and run queries

import psycopg
import boto3
import os, sys

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

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

    # connection parameters
    dbname = "dbname=postgres"
    user = "user=admin"
    host = f'host={cluster_endpoint}'
    sslmode = "sslmode=verify-full"
    sslrootcert = "sslrootcert=system"
    password = f'password={password_token}'

    # Make a connection to the cluster
    conn = psycopg.connect('%s %s %s %s %s %s' % (dbname, user, host, sslmode, sslrootcert, password))

    conn.set_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()
    
    # Verify that the result we got is what we inserted before
    assert row[0] != None
    assert row[1] == "John Doe"
    assert row[2] == "Anytown"
    assert row[3] == "555-555-1999"
    
    # 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 = "foo0bar1baz2quux3quuux4.dsql.us-east-1.on.aws"
    main(cluster_endpoint)