PythonでAWS Lambdaの関数を作ろうして調べたことまとめ

はじめに

今までPythonにはまったく触れずに来たのですが、お仕事でPythonを使ってAWS Lambda Functionの処理を組む機会があったので、あれこれ調べたことを備忘録代わりにまとめてみました。
現状もまだPython初心者のままですので、残念なところがあったらご指摘いただけるとうれしいです

実行環境

• Python 3.8 （Lambdaで実行するときのランタイムも同バージョン）
  • pytest 6.2.1
  • pg8000 1.16.6
  • moto 1.3.16

ログ出力にリクエストIDを含めたい

Lambdaでは標準出力の内容がCloudWatch上で確認できますが、そのままでは運用には少々厳しい。せめてSTARTとENDのときに出力されるリクエストIDを添えて出力したい。

対応方法：loggingを使う

公式ドキュメントにも記載がありました。 ログフォーマットの変更方法とextraの設定方法まで調べてからこのドキュメントを見つけて泣いた 特に気にせずloggerを用意するだけでよかったんですね。。。

import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

logger.info('test!')

出力結果

[INFO]  2020-01-31T22:12:58.534Z    1c8df7d3-xmpl-46da-9778-518e6eca8125    test!

CloudWatchでログをフィルタリングしたい

ログ出力にリクエストIDを含むようになったので、今度はそのリクエストIDでログをフィルタリングしたい。1リクエスト分のログが追いやすくなる。

調べたところ、JSON形式のログであればスマートにできそう。
例えば aws_request_id = "1c8df7d3-xmpl-46da-9778-518e6eca8125" の行だけフィルタして表示したい場合、フィルタ文字列に以下のように入力してやるとそれが叶う。

{$.aws_request_id = "1c8df7d3-xmpl-46da-9778-518e6eca8125"}

では、ログをJSON形式で出力するにはどうしたらいいか。

解決方法：Loggingのフォーマッタを変更する

いくつかライブラリがGitHubにも公開されていましたが、ひとまずは手作りで、というか先人の記事を参考にさせていただきました。最初から機能モリモリにしなくてもこれで不都合が発生したらJsonFormatterを改造するなり改めてライブラリを探すなりしたらいいかな、というさじ加減です。
参考： https://qiita.com/tonluqclml/items/780370a4575781eb19df

import json
import logging

class JsonFormatter:
    def format(self, record):
        return json.dumps(vars(record))

for handler in logging.getLogger().handlers:
    handler.setFormatter(JsonFormatter())
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
logger.info('hoge')

出力結果（一部（だいぶ）省略）

{
    "name": "sample",
    "msg": "hoge",
    "args": [],
    "levelname": "INFO",
    "levelno": 20,
    "funcName": "lambda_handler",
    "threadName": "MainThread",
    "processName": "MainProcess",
    "aws_request_id": "1c8df7d3-xmpl-46da-9778-518e6eca8125"
}

テストでAWSの機能をモックにしたい

解決方法：motoを使う

テストメソッドに @mock_s3 などモック化したいモジュールのアノテーションを定義するだけ。かんたんで感動しました。
いざ実装する際はモックの振る舞いを定義するにはどうしたら？と思ったのですが、ふつうにboto3を使ってAWSのリソースを操作するコードをテストの前準備（S3にファイルをアップロードしておくとか、SSMに任意のパラメータを定義しておくとか）として書いておけばよいみたいです。なるほどなー。

@mock_s3
@mock_ssm
def test_handler_S3オブジェクトの内容が結果として返ること():

    # ssmのパラメータストアに値を設定
    ssm = boto3.client('ssm')
    ssm.put_parameter(
        Name='/test/s3import/objectkey',
        Description='test key',
        Value='test/object/key/data.csv',
        Type="String",
    )
    ssm.put_parameter(
        Name='/test/s3import/s3bucket',
        Description='test bucket',
        Value='testBucketName',
        Type="String",
    )

    # 環境変数の設定
    os.environ['AWS_DEFAULT_REGION'] = 'us-east-2'

    # s3 バケットを作成
    s3 = boto3.resource('s3')
    s3_bucket = s3.Bucket('testBucketName')
    s3_bucket.create(CreateBucketConfiguration={'LocationConstraint': 'us-west-1'})
    # ファイルをアップロード
    buf = BytesIO()
    with gzip.open(buf, mode="wt") as f:
        f.write('1,2,3')
    s3_bucket.put_object(Body=buf.getvalue(), Key='test/object/key/data.csv')

    # テスト実行
    actual = app.handler({}, {})

    # 検証
    assert actual == { 'item1': 1, 'item2': 2, 'item3': 3 }

テストでDBを差し替えたい

これだけAWSあんまり関係ないんですが、RDSを使うLambdaを実装することもあろう、ということで。
javaではH2 Databaseなどを使ってテストケースを実行していたのですが、Pythonでも似たようなことはできないのかなと思って調べてみました。

解決方法（？）： SQLiteを使う

たまたま使っていた pg8000 というpostgresql用のライブラリと、Pythonが標準で用意してくれているSQLiteのインターフェイスである sqlite3 のインターフェイスが似ているというだけで採用してみました。。。
pg8000もバージョンアップしていて今はインターフェイスが変わっているみたいなので、そちらにあわせるともう少し考えないとだめだろうと思いますが、標準SQLを使ったクエリであればこれでも動作はすると思います。ただし、もっと細かい型による挙動のちがい（特に日付型など）や、postgresql独特のSQLの文法やオブジェクトを使った処理がある場合、この手段は使えませんのでご注意を（H2でも似たようなものですが）。

なお、SQLiteはインメモリ指定にしてしまうとコネクションをクローズする処理が呼び出された時点でDBは消えてしまうようなので、 tempfile を使ってDBを用意することにしました。これならテスト実行後にファイルの削除処理などを書かなくてよいのでらくちん。

db_file = tempfile.NamedTemporaryFile().name

@mock.patch('pg8000.connect', return_value = sqlite3.connect(db_file))
def test_handler_DBから取得した値が返ること(conn_mock):

    with sqlite3.connect(db_file) as conn:
        with closing(conn.cursor()) as cur:
            cur.execute('CREATE TABLE test_tbl (column1 character(1), column2 character(1), PRIMARY KEY (column1)) ')
            cur.execute('INSERT INTO test_tbl (column1, column2) VALUES ("1", "2")')
 
    # テスト実行
    actual = app.handler({}, {})

    # 検証
    assert len(actual) == 1
    assert actual[0] == { 'column1': '1', 'column2': '2' }

可能であれば、Dockerでpostgresqlのコンテナを立ち上げるとか、テスト実行のときだけ小さいRDSのインスタンスをたてちゃうとか、ちゃんとしたpostgresqlを用意したほうがよりきちんとした検証ができると思います。今のところ、見つけられたお手軽な手段がsqlite3だったというだけなので、他にもよいものがあったら試していきたいです。

おわりに

Pythonを実行するためにはどうしたら？？？というところから始めたので未だに変なところでハマったりしますが（自作ライブラリのimportとか…）、標準のライブラリだけでも思ったよりいろんなことができるんだなと感じました。せっかくなのでもうちょっとPython書けるようになりたいな〜。フレームワークを使った処理なども書いてみたいです。