使用 IAM 身分驗證和 AWS SDK for Python (Boto3) 連線至資料庫叢集

您可以如下所述使用 AWS SDK for Python (Boto3)，連線至 Aurora MySQL 或 Aurora PostgreSQL 資料庫叢集。

先決條件

以下是使用 IAM 身分驗證連線至資料庫叢集的先決條件：

• 啟用和停用 IAM 資料庫身分驗證

• 建立並使用 IAM 政策進行 IAM 資料庫存取

• 使用 IAM 身分驗證建立資料庫帳戶

此外，請確定範例程式碼中匯入的程式庫存在於您的系統上。

範例

程式碼範例使用設定檔進行共用登入資料。如需指定登入資料的相關資訊，請參閱 AWS SDK for Python (Boto3) 文件中的登入資料。

以下程式碼範例顯示如何產生身分驗證字符，然後用來連線至資料庫叢集。

若要執行此程式碼範例，您必須從 AWS SDK for Python (Boto3) 網站上取得 AWS。

視需要修改下列變數的值：

• ENDPOINT – 您想要存取之資料庫叢集的端點

• PORT – 用於連線資料庫叢集的連接埠號碼

• USER – 您想要存取的資料庫帳戶

• REGION – 執行資料庫叢集的 AWS 區域

• DBNAME – 您想要存取的資料庫

• SSLCERTIFICATE – Amazon Aurora 之 SSL 憑證的完整路徑

  對於 ssl_ca，指定 SSL 憑證。若要下載 SSL 憑證，請參閱使用 SSL/TLS 加密資料庫叢集叢集的連線。

注意

您無法使用自訂 Route 53 DNS 記錄或 Aurora 自訂端點替代資料庫叢集端點來產生身分驗證字符。

此程式碼連接到一個 Aurora MySQL 資料庫叢集。

在執行此程式碼之前，請遵循 Python 套件索引中的指示，來安裝 PyMySQL 驅動程式。

import pymysql
import sys
import boto3
import os

ENDPOINT="mysqlcluster.cluster-123456789012.us-east-1.rds.amazonaws.com"
PORT="3306"
USER="jane_doe"
REGION="us-east-1"
DBNAME="mydb"
os.environ['LIBMYSQL_ENABLE_CLEARTEXT_PLUGIN'] = '1'

#gets the credentials from .aws/credentials
session = boto3.Session(profile_name='default')
client = session.client('rds')

token = client.generate_db_auth_token(DBHostname=ENDPOINT, Port=PORT, DBUsername=USER, Region=REGION)

try:
    conn =  pymysql.connect(host=ENDPOINT, user=USER, passwd=token, port=PORT, database=DBNAME, ssl_ca='SSLCERTIFICATE')
    cur = conn.cursor()
    cur.execute("""SELECT now()""")
    query_results = cur.fetchall()
    print(query_results)
except Exception as e:
    print("Database connection failed due to {}".format(e))          
                

此程式碼連接到一個 Aurora PostgreSQL 資料庫叢集。

在執行此程式碼之前，請按照 Psycopg 文件中的指示來安裝 psycopg2。

import psycopg2
import sys
import boto3
import os

ENDPOINT="postgresmycluster.cluster-123456789012.us-east-1.rds.amazonaws.com"
PORT="5432"
USER="jane_doe"
REGION="us-east-1"
DBNAME="mydb"

#gets the credentials from .aws/credentials
session = boto3.Session(profile_name='RDSCreds')
client = session.client('rds')

token = client.generate_db_auth_token(DBHostname=ENDPOINT, Port=PORT, DBUsername=USER, Region=REGION)

try:
    conn = psycopg2.connect(host=ENDPOINT, port=PORT, database=DBNAME, user=USER, password=token, sslrootcert="SSLCERTIFICATE")
    cur = conn.cursor()
    cur.execute("""SELECT now()""")
    query_results = cur.fetchall()
    print(query_results)
except Exception as e:
    print("Database connection failed due to {}".format(e))                
                

如果要透過 Proxy 連線到資料庫叢集，請參閱 使用 IAM 身分驗證連線到代理。