IAM 認証および AWS SDK for Python (Boto3) を使用した DB インスタンスへの接続

次に説明するように、AWS SDK for Python (Boto3) を使用して、RDS for MariaDB、MySQL、または PostgreSQL DB インスタンスに接続できます。

前提条件

IAM 認証を使用して DB インスタンスに接続するための前提条件は以下のとおりです。

さらに、サンプルコード内のインポートされるライブラリがシステムに存在することを確認してください。

例

コード例では、共有認証情報のプロファイルを使用します。認証情報の指定については、AWS SDK for Python (Boto3) ドキュメントの「認証情報」を参照してください。

以下のコード例で、認証トークンを生成し、それを使用して DB インスタンスに接続する方法を示します。

このコードサンプルを実行するには、AWS SDK for Python (Boto3) サイトにある AWS が必要です。

必要に応じて以下の可変の値を変更します。

このコードで MariaDB インスタンスまたは MySQL DB インスタンスに接続します。

このコードを実行する前に、Python Package Index の手順に従って 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(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 DB インスタンスに接続します。

このコードを実行する前に、Psycopg documentation の手順に従って 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))                
                

プロキシ経由で DB インスタンスに接続する場合は、「IAM 認証を使用したプロキシへの接続」を参照してください。