使用 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
– 要访问的数据库 -
SSLCERTIFICATE
– Amazon RDS 的 SSL 证书的完整路径对于
ssl_ca
,请指定 SSL 证书。要下载 SSL 证书,请参阅 使用 SSL/TLS 加密与数据库实例或集群的连接。
注意
您不能使用自定义 Route 53 DNS 记录代替数据库实例端点来生成身份验证令牌。
此代码连接到 MariaDB 或 MySQL 数据库实例。
在运行此代码之前,请按照 Python 包索引
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(auth_plugin_map={'mysql_clear_password':None},host=ENDPOINT, user=USER, password=token, port=PORT, database=DBNAME, ssl_ca='SSLCERTIFICATE
', ssl_verify_identity=True) 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 身份验证连接到代理。