Function Compute:RDS MySQL にアクセスするためのベストプラクティス

ステップ 2: 関数で RDS にアクセスする

1. Function Compute コンソール にログインし、関数リストでターゲット関数を見つけ、関数の詳細ページで [コード] タブをクリックし、コードエディターで次のサンプルコードを記述します。

   from flask import Flask, jsonify
   import pymysql
   import os
   from datetime import datetime
   import logging
   
   app = Flask(__name__)
   
   # MySQL シングルトン接続を格納するグローバル変数
   _mysql_connection = None
   
   # データベース接続を作成する（シングルトンパターン）
   def getConnection():
       global _mysql_connection
       try:
           # 接続が既に存在し、切断されていない場合は、それを直接返す
           if _mysql_connection is not None:
               try:
                   # 接続が有効かどうかをテストする（簡単なコマンドテスト）
                   with _mysql_connection.cursor() as cursor:
                       cursor.execute("SELECT 1")  # 接続状態をテストするための簡単なクエリ
                       result = cursor.fetchone()
                       if result and result[0] == 1:
                           return _mysql_connection
               except pymysql.OperationalError:
                   # 接続が切断されている場合は、リセットする
                   _mysql_connection = None
   
           # 接続が存在しないか、切断されている場合は、再作成する
           _mysql_connection = pymysql.connect(
               host=os.environ['MYSQL_HOST'],
               port=int(os.environ['MYSQL_PORT']),
               user=os.environ['MYSQL_USER'],
               password=os.environ['MYSQL_PASSWORD'],
               db=os.environ['MYSQL_DBNAME']
           )
           return _mysql_connection
       except Exception as e:
           logging.error(f"データベース接続中にエラーが発生しました: {e}")
           raise
   
   
   @app.route('/', defaults={'path': ''})
   @app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
   def hello_world(path):
       conn = getConnection()
       try:
           with conn.cursor() as cursor:
               # users テーブルのすべてのレコードをクエリする。users テーブルは実際のテーブル名に応じて変更する
               sql = "SELECT * FROM users"
               cursor.execute(sql)
               result = cursor.fetchall()
               
               columns = [desc[0] for desc in cursor.description]  # フィールド名リストを取得する
   
               # クエリ結果を辞書リストに変換する
               users = []
               for row in result:
                   user = {}
                   for idx, column_name in enumerate(columns):
                       value = row[idx]
                       if isinstance(value, datetime):  # 日付タイプのフィールドを処理する
                           user[column_name] = value.strftime('%Y-%m-%d %H:%M:%S')
                       else:
                           user[column_name] = value
                   users.append(user)
                   
               if users:
                   # すべてのユーザーの JSON レスポンスを返す
                   return jsonify(users), 200
               else:
                   # ユーザーが見つからない場合は、404 エラーを返す
                   return jsonify({'error': 'ユーザーが見つかりません'}), 404
   
       except Exception as e:
           logging.error(f"データベース操作中にエラーが発生しました: {e}")
           return jsonify({'error': 'データベースエラー'}), 500
   
   
   if __name__ == '__main__':
       app.run(host='0.0.0.0', port=9000)
   
   

2. 関数の詳細ページで、[構成] > [環境変数] を選択し、[編集] をクリックして、[環境変数] パネルで次の環境変数を構成します。

   環境変数名	環境変数値	説明
   MYSQL_HOST	rm-bp19u8e76ae****.mysql.rds.aliyuncs.com	データベースインスタンスのエンドポイント。 同じ VPC 内の RDS データベース シナリオを選択した場合は、この環境変数をデータベースの内部エンドポイントに設定します。 VPC またはリージョンを跨いで RDS データベースにアクセスする シナリオを選択した場合は、この環境変数をデータベースのパブリックエンドポイントに設定します。 RDS インスタンスリスト で、ターゲット RDS インスタンス ID をクリックします。左側のナビゲーションウィンドウで、[データベース接続] をクリックします。[データベース接続] ページで、データベースの内部エンドポイントまたはパブリックエンドポイントを取得します。
   MYSQL_DBNAME	db_test	RDS インスタンスに作成されたデータベースの名前。
   MYSQL_PASSWORD	*****	データベースアカウントのパスワード。
   MYSQL_PORT	3306	データベースインスタンスのプライベートポート。
   MYSQL_USER	dms_user_****	RDS インスタンスへの接続に使用するアカウントのユーザー名。

3. 関数の詳細ページで、[コード] タブをクリックし、[関数のテスト] をクリックして、正常に実行された後に返された結果を表示します。テーブルクエリ操作は正常に完了しました。