- Pythonで踏み台サーバー経由でDB接続する方法についてメモする。
環境構成
- ローカルPCから踏み台サーバー(EC2)を経由し、DB(Aurora)に接続するようなケースを想定
コード
-
conf/setting.conf- 踏み台サーバーとDBの接続情報を記述する。
# 踏み台サーバー接続情報 [BASTION] BASTION_HOST = YOUR_BASTION_HOST_IP BASTION_PORT = 22 BASTION_HOST_KEY = None BASTION_USERNAME = YOUR_BASTION_USERNAME BASTION_PASSWORD = YOUR_BASTION_PASSWORD BASTION_PKEY = None BASTION_REMOTE_BIND_ADDRESS_HOST = YOUR_AURORA_HOST BASTION_REMOTE_BIND_ADDRESS_PORT= 3306 # DB接続情報 [DB] DB_USER = YOUR_DB_USERNAME DB_PASSWORD = YOUR_DB_PASSWORD DB_NAME = YOUR_DB_NAME -
test.py- 設定ファイル
conf/setting.confから取得した接続情報を用いてDB接続を行い、SELECT文を実行して、実行結果をresultフォルダ配下に保存する。
- 設定ファイル
import sshtunnel
from sqlalchemy import create_engine
import pandas as pd
import configparser
# 設定ファイルから接続情報読み込み
conf = configparser.ConfigParser()
conf.read('conf/setting.conf', encoding='utf-8')
# 踏み台サーバー接続情報
BASTION_HOST = conf['BASTION']['BASTION_HOST']
BASTION_PORT = int(conf['BASTION']['BASTION_PORT'])
if conf['BASTION']['BASTION_HOST_KEY'] == 'None':
BASTION_HOST_KEY = None
else:
BASTION_HOST_KEY = conf['BASTION']['BASTION_HOST_KEY']
BASTION_USERNAME = conf['BASTION']['BASTION_USERNAME']
BASTION_PASSWORD = conf['BASTION']['BASTION_PASSWORD']
ifconf['BASTION']['BASTION_PKEY'] == 'None':
BASTION_PKEY = None
else:
BASTION_PKEY = conf['BASTION']['BASTION_PKEY']
BASTION_REMOTE_BIND_ADDRESS_HOST = conf['BASTION']['BASTION_REMOTE_BIND_ADDRESS_HOST']
BASTION_REMOTE_BIND_ADDRESS_PORT = int(conf['BASTION']['BASTION_REMOTE_BIND_ADDRESS_PORT'])
# DB接続情報
DB_USER = conf['DB']['DB_USER']
DB_PASSWORD = conf['DB']['DB_PASSWORD']
DB_NAME = conf['DB']['DB_NAME']
# DB接続
sshtunnel.SSH_TIMEOUT = 5.0
sshtunnel.TUNNEL_TIMEOUT= 5.0
with sshtunnel.SSHTunnelForwarder((BASTION_HOST, BASTION_PORT), ssh_host_key=BASTION_HOST_KEY,
ssh_username=BASTION_USERNAME, ssh_password=BASTION_PASSWORD,
ssh_pkey=BASTION_PKEY,
remote_bind_address=(BASTION_REMOTE_BIND_ADDRESS_HOST, BASTION_REMOTE_BIND_ADDRESS_PORT),local_bind_address=('localhost',3306)) as server:
con = create_engine(
f'mysql+pymysql://{DB_USER}:{DB_PASSWORD}@localhost:{server.local_bind_port}/{DB_NAME}?charset=utf8',
echo=True)
# pandasでSQL実行+実行結果を取得
df = pd.read_sql('SELECT * FROM fuga', con)
# CSV 保存
df.to_csv('result/data.csv')
# Excel 保存
# df.to_excel('result/data.xlsx',sheet_name='data',index=False)
実行方法
python test.py
※resultディレクトリにSELECT文実行結果が保存される。