使用 IAM 身份验证和 Amazon SDK for Python (Boto3) 连接到数据库实例 - Amazon Relational Database Service
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

使用 IAM 身份验证和 Amazon SDK for Python (Boto3) 连接到数据库实例

您可以使用 Amazon SDK for Python (Boto3) 连接到 RRDS for MariaDB、MySQL 或 PostgreSQL 数据库实例 ,如下所述。

先决条件

以下是使用 IAM 身份验证连接到数据库实例的先决条件:

此外,请确保示例代码中的导入库存在于您的系统中。

示例

该代码示例将配置文件用于共享凭证。有关指定凭证的信息,请参阅 Amazon SDK for Python (Boto3) 文档中的凭证

以下代码示例演示如何生成身份验证令牌,然后使用该令牌连接到数据库实例

要运行该代码示例,您需要使用 Amazon SDK for Python (Boto3) 网站上提供的 Amazon

根据需要修改以下变量的值:

  • ENDPOINT – 要访问的数据库实例的终端节点。

  • PORT – 用于连接到数据库实例的端口号

  • USER – 要访问的数据库账户

  • REGION – 在其中运行数据库实例的 Amazon 区域

  • DBNAME – 要访问的数据库

  • SSLCERTIFICATEAmazon RDS 的 SSL 证书的完整路径

    对于 ssl_ca,请指定 SSL 证书。要下载 SSL 证书,请参阅 使用 SSL/TLS 加密与数据库实例的连接

注意

您不能使用自定义 Route 53 DNS 记录代替数据库实例端点来生成身份验证令牌。

此代码连接到 MariaDB 或 MySQL 数据库实例。

在运行此代码之前,请按照 Python 包索引中的说明安装 PyMySQL 驱动程序。

import pymysql import sys import boto3 import os ENDPOINT="mysqldb.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))

此代码连接到 PostgreSQL 数据库实例。

在运行此代码之前,请按照 Psycopg 文档中的说明安装 psycopg2

import psycopg2 import sys import boto3 import os ENDPOINT="postgresmydb.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 身份验证连接到代理