使用 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))                
                

如果您想通过代理连接到数据库集群，请参阅使用 IAM 身份验证连接到代理。