db_access_by_lambda.py

db_access_by_lambda.py

# $ python3 -m venv env
# $ source env/bin/activate
# $ mkdir mysql_test;cd mysql_test
# $ pip install mysqlclient -t ./
# mysqlclient is needs "gcc, python-devel, libmysqlclient-devel"
# $ vi lambda_function.py
# $ zip -r9 lambda_mysql.zip .
# zipfile upload to aws.
# It may be necessary to set up vpc or subnet

import MySQLdb
import json
from datetime import datetime

def lambda_handler(event, context):
    # DB続情報
    db_config = {
        'host': 'example.com',
        'db': 'test_db',
        'user': 'test_user',
        'passwd': 'test_password',
        'charset': 'utf8',
    }
    try:
        # 接続
        connect = MySQLdb.connect(
                            host=db_config['host'],
                            db=db_config['db'],
                            user=db_config['user'],
                            passwd=db_config['passwd'],
                            charset=db_config['charset']
                        )

        # SQL実行用カーソル取得
        connect.cursorclass = MySQLdb.cursors.DictCursor
        cursor = connect.cursor()

        sql = 'SELECT * FROM test_table WHERE test_col IN (%s,%s,%s) limit %s'
        sql_param = [1, 2, 3, 10]

        # SQL実行
        cursor.execute(sql, sql_param)
        #print(cursor.rowcount) # 件数取得
        #print(cursor.fetchall()[0]) # 全件取得
        #print(cursor._last_executed.decode('utf-8')) # 直前に実行したSQL取得(decodeしないと文字化け)
        
        res = {
            "rowcount": cursor.rowcount,
            "rows": cursor.fetchall()[0]
        }

        # 切断
        connect.close()
        
        return json.dumps(res, default=support_datetime_default)
        
    except MySQLdb.Error as ex:
        print('MySQL Error: ', ex)

def support_datetime_default(o):
    if isinstance(o, datetime):
        return o.isoformat()
    raise TypeError(repr(o) + " is not JSON serializable")