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 색인에 있는 지침에 따라 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 DB 인스턴스에 연결됩니다.

이 코드를 실행하기 전에 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))                
                

프록시를 통해 DB 인스턴스에 연결하려는 경우 IAM 인증을 사용하여 프록시에 연결을 참조하세요.