データベースを利用するアプリケーション開発において、SQLクエリの管理と実行は重要な課題です。
特に、AWS Lambdaのようなサーバーレス環境で開発を行う場合、効率的な開発と検証プロセスが求められます。
本記事では、PythonとTiDB Serverlessを使ってSQLクエリを管理し、
ローカル環境での検証とAWS Lambda上での実行を効率化する方法を紹介します。
TiDB Serverlessは、MySQLと互換性のあるサーバーレスのデータベースです。
DBサーバーを自分で構築する手間が省けるので、検証には最適と言えます。
検証であれば、無料で使える枠が用意されています。
ローカル環境での開発とSQL検証の重要性
AWS Lambda上で直接開発を行うことは、以下のような理由から非効率的です。
- Lambda関数の実行にはコストがかかる
- Lambda上での開発ツールは、ローカル環境ほど充実していない
- Lambda上でのデバッグとテストは時間がかかる
そのため、SQLクエリの検証はローカル環境で行うことをお勧めします。
ローカル環境では、充実した開発ツールを使ってクエリの動作を確認し、必要な修正を加えることができます。
Lambda関数にも無料枠はあります。
しかし、工夫をしなければLambda関数上での開発は非効率です。
コードとSQLの分離によるメリット
本事例では、SQLクエリをPythonコードから分離し、combined_queries.sqlというファイルで管理しています。
このファイルを用いた開発方法は、次の記事をご覧ください。
この方法には、以下のようなメリットがあります。
- SQLクエリの再利用性が向上する
- コードの可読性とメンテナンス性が向上する
- ローカル環境とLambda関数間でのクエリの共有が容易になる
ローカル環境で検証したSQLクエリを、Lambda関数にデプロイする際は、combined_queries.sqlファイルを更新するだけで済みます。
これにより、開発と検証のプロセスが大幅に効率化されます。
ローカル環境でのSQL検証
ローカル環境でSQLクエリを検証するには、以下のPythonコードを使用します。
import pymysql
import time
from config import Config
config = Config()
db_config = {
'host': config.tidb_host,
'port': 4000,
'user': config.tidb_user,
'password': config.tidb_password,
'database': config.tidb_db_name,
'ssl_verify_cert': True,
'ssl_verify_identity': True,
'ssl_ca': config.ca_path
}
def get_query(query_name):
with open('combined_queries.sql', 'r', encoding='utf-8') as file:
content = file.read()
query_start = content.find(f'-- [{query_name}]\n')
if query_start == -1:
raise ValueError(f"Query '{query_name}' not found.")
query_end = content.find('-- [', query_start + 1)
if query_end == -1:
query_end = len(content)
query_text = content[query_start:query_end].strip()
query_lines = query_text.split('\n')
query = query_lines[-1].strip()
return query
def execute_query(query_name, params):
query = get_query(query_name)
query = query.replace('?', '%s') # プレースホルダーを%sに変換
connection = pymysql.connect(**db_config)
cursor = connection.cursor()
start_time = time.perf_counter()
cursor.execute(query, params)
data = cursor.fetchall()
end_time = time.perf_counter()
cursor.close()
connection.close()
return data, end_time - start_time
def main():
city_id = 300
data, execution_time = execute_query('get_city_data_by_id', (city_id,))
print(f"Result of get_city_data_by_id (city_id {city_id}): {data}")
print(f"Execution time: {execution_time:.5f} seconds")
print()
if __name__ == '__main__':
main()
get_query関数でcombined_queries.sqlからクエリを取得し、execute_query関数でTiDBに接続してクエリを実行しています。
ローカル環境では、TiDBの接続情報をdb_config変数で管理する形となります。
具体的には、.envファイルに設定情報を登録しています。
その.envをpython-dotenvで読み込んでいるということです。
このコードを使って、ローカル環境でSQLクエリの動作を確認し、必要な修正を加えます。
なお、データにはMySQLのサンプルデータベースである「world database」を利用しています。
TiDB Serverlessの管理画面で次のように確認できます。
Lambda関数へのデプロイ
ローカル環境で検証したSQLクエリをLambda関数で使用するには、以下の手順を実行します。
- combined_queries.sqlファイルをLambda関数のデプロイパッケージに含める
- Lambda関数のコードを更新する
Lambda関数のコードは、以下のようになります。
import pymysql
import time
import os
def get_query(query_name):
with open('combined_queries.sql', 'r', encoding='utf-8') as file:
content = file.read()
query_start = content.find(f'-- [{query_name}]\n')
if query_start == -1:
raise ValueError(f"Query '{query_name}' not found.")
query_end = content.find('-- [', query_start + 1)
if query_end == -1:
query_end = len(content)
query_text = content[query_start:query_end].strip()
query_lines = query_text.split('\n')
query = query_lines[-1].strip()
return query
def execute_query(query_name, params):
query = get_query(query_name)
query = query.replace('?', '%s') # プレースホルダーを%sに変換
db_config = {
'host': os.environ['TIDB_HOST'],
'port': int(os.environ['TIDB_PORT']),
'user': os.environ['TIDB_USER'],
'password': os.environ['TIDB_PASSWORD'],
'database': os.environ['TIDB_DB_NAME'],
'ssl_verify_cert': True,
'ssl_verify_identity': True,
'ssl_ca': os.environ['CA_PATH']
}
connection = pymysql.connect(**db_config)
cursor = connection.cursor()
start_time = time.perf_counter()
cursor.execute(query, params)
data = cursor.fetchall()
end_time = time.perf_counter()
cursor.close()
connection.close()
return data, end_time - start_time
def lambda_handler(event, context):
city_id = 300
data, execution_time = execute_query('get_city_data_by_id', (city_id,))
print(f"Result of get_city_data_by_id (city_id {city_id}): {data}")
print(f"Execution time: {execution_time:.5f} seconds")
return {
'statusCode': 200,
'body': 'Query executed successfully'
}
Lambda関数では、TiDBの接続情報を環境変数から取得しています。
また、TiDBへの接続にはPyMySQLライブラリを利用しています。
AWS LambdaへのPythonライブラリのインストールは、以下の記事が参考になります。
まとめ
PythonとTiDBを使ってSQLクエリを管理する方法を紹介しました。
SQLクエリをコードから分離することで、再利用性と可読性が向上します。
その結果、ローカル環境とLambda関数間でのクエリの共有が容易になります。
また、ローカル環境でSQLクエリを検証することで、開発と検証のプロセスを大幅に効率化できます。
本事例の方法を採用することで、AWS Lambdaを使ったデータベースアプリケーションの開発と運用の生産性が向上するでしょう。
ちなみに、各環境における検証コードの実行結果は以下となります。
ローカルマシン
Result of get_city_data_by_id (city_id 300): ((300, 'Gravataí', 'BRA', 'Rio Grande do Sul', 223011),) Execution time: 0.01424 seconds
Lambda関数
Result of get_city_data_by_id (city_id 300): ((300, 'Gravataí', 'BRA', 'Rio Grande do Sul', 223011),) Execution time: 0.00859 seconds
Lambda関数は、かなり高速です。
TiDB ServerlessはAWSの東京リージョンで動いていることが判明しています。
そして、Lambda関数も東京リージョンで動かしています。
つまり、TiDB ServerlessとLambda関数はかなり近所である可能性はありますね。
