SQLAlchemyはPythonのORM(Object Relational Mapper)ライブラリーです。
SQLAlchemyはDBに接続する際、標準でコネクションプーリング機能が組み込まれています。
pythonのWebフレームワークであるFlaskからDBにアクセスする際、毎回毎回DBの接続をしていては時間がかかるので、このSQLAlchemyのコネクションプールを利用すれば2回目以降の接続が速くなります。SQLAlchemyのDB接続はORMなしでも使用可能です。
当記事はORMは使わず、SQLAlchemyのDB接続機能のみを使用して、FlaskからDb2にアクセスし、SQLを実行する方法について説明します。
1. 前準備
SQLAlchemyでDb2に接続するには以下のライブラリが必要です：
pipでインストールします：
pip install ibm_db
pip install ibm_db_sa
pip install sqlalchemy
尚、当記事ではFlaskからアクセスしますので、Flaskのライブラリも必要です。
2. SQLAlchemyのDb2接続URL
以下になります。（{}はDbの接続情報で置き換えます):
ibm_db_sa://{username}:{password}@{hostname}:{port}/{database};
SSL接続の場合は最後にSECURITY=SSL;を付加します。
ibm_db_sa://{username}:{password}@{hostname}:{port}/{database};SECURITY=SSL;
passwordに＠などの特殊文字が入っていると正しく動作しません。このような場合はURLエンコーディングします。
参考： Escaping Special Characters such as @ signs in Passwords
以下は、パスワード「kx@jj5/g」を含むURLの例です。「at」記号とスラッシュ文字は、それぞれ%40と%2Fとして表現されます：
from urllib.parse import quote_plus
password = quote_plus(("kx@jj5/g")
print(password)
出力:
kx%40jj5%2Fg
3. SQLAlchemyのエンジンを作成
engine = create_engine(url)でSQLAlchemyのエンジンを作成できます。
何もパラメータを指定しないと、デフォルトでコネクションプールが作成されます。
コネクションプールの種類はデフォルトがQueuePoolです。 変更したい場合はcreate_engine（）のパラメーターpoolclassで指定できます。
以下のクラスメソッド株式会社さんの記事に指定可能なコネクションプールの種類およびパラメーターや動作などついて詳細な説明がありましたので参考にしてみてください:
ここではデフォルトのQueuePoolを使用します。
デフォルトでは5つの接続を永続化してプールします。さらに追加でデフォルトでは10接続オープン可能です。
このあたりの数の設定はpool_sizeとmax_overflowというcreate_engine（）のパラメーターで行えます。詳細の説明は上記Linkにありました。ここでは何も指定せずデフォルト設定を使用しています。
以下がサンプルコードです。
from sqlalchemy import create_engine
from urllib.parse import quote_plus
# DB2の接続情報を設定
database = "xxxx"
hostname = "xxxxx.xxx.xxxx.xxx.xxx"
port = "xxxxx"
username = "xxxxxx"
password = quote_plus("@12345mko@12345mko0@Zaq2wsx")
# SQLAlchemyのエンジンを作成
url = f"ibm_db_sa://{username}:{password}@{hostname}:{port}/{database};SECURITY=SSL;"
engine = create_engine(url)
4. Db接続とSQLの実行
3で取得したengineを使用します。
Db接続はengine.connect() で取得できます。コネクションプールに使用していない接続があればそれを利用します。何もなければ新規作成されます。
connection = engine.connect()
SQLは上記で取得したconnectionのexecuteメソッドで実行できます。
DB接続と合わせるとこんな感じです: ここから引用
from sqlalchemy import text
with engine.connect() as connection:
result = connection.execute(text("select username from users"))
for row in result:
print("username:", row["username"])
5. Flaskでの考慮点と実装サンプル
engineはdisposeしないと接続が完全に切断されません。
FlaskだとCtrl+Cで落としてしまうことが多いので、atexitで終了時に実行する関数を登録しておくと安全です。 以下のような感じです。
import atexit
# 終了時に実行する関数
def close_process():
engine.dispose() #プールしたDB接続の切断処理
# 終了時に実行する関数を登録
atexit.register(close_process)
ということで、ここまでをまとめてFlaskで実装したサンプルが以下になります。
単にルートにアクセスするとベタ打ちでSQLの結果を返します(SQLエラーハンドリング等は未実装)。
from flask import Flask
from sqlalchemy import create_engine, text
from urllib.parse import quote_plus
import atexit
engine = None
# 終了時に実行する関数
def close_process():
engine.dispose() #プールしたDB接続の切断処理
app = Flask(__name__)
# DB2の接続情報を設定
database = "xxxx"
hostname = "xxxxx.xxx.xxxx.xxx.xxx"
port = "xxxxx"
username = "xxxxxx"
password = quote_plus("@12345mko@12345mko0@Zaq2wsx")
# SQLAlchemyのエンジンを作成(SSL接続)
url = f"ibm_db_sa://{username}:{password}@{hostname}:{port}/{database};SECURITY=SSL;"
engine = create_engine(url)
# 終了時に実行する関数を登録
atexit.register(close_process)
@app.route('/')
def index():
with engine.connect() as connection:
result = connection.execute(text("select username from users"))
rows = result.fetchall()
return str(rows) #ベタ打ちで結果を返す
if __name__ == '__main__':
app.run(debug=True)
これを実行してブラウザーでアクセスすると、初回アクセスは実際にDBに接続するので遅いですが、2回目以降DB接続再利用の場合は速くなります。
SQLAlchemyでORMを使用しない細かい使用方法は公式のWorking with Engines and Connectionsあたりから確認するのがよいと思います。SQLAlchemyは本来はORMでバリバリ使うべきなのかもしれませんが、コネクションプールだけ使いたい場合に便利です。
以上です。