SQLクエリの管理と実行をPythonで効率化 – AWS Lambdaの活用事例

データベースを利用するアプリケーション開発において、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関数はかなり近所である可能性はありますね。