AWS LambdaをJavaで書いていた人がPythonで書くためにすること

こんにちは。ゴーリスト開発の盛次です。

Javaが大好きだけどPythonでもAWS Lambdaを書いてみるかー、と思った人のための記事です。

開発環境

Windows10
PyCharm
Python3.6

開発環境を作る

とりあえずインストールが楽そうなのでAnacondaを利用します。
Anacondaのインストール方法は世の中に優れた記事がたくさんあるのでそちらにお任せします。
JetBrains製品が大好きなのでIDEはPyCharmを使います。
CommunityとProfessionalの２種類がありますが、無料のCommunityで問題ないです。リモートデバッグを使う場合はProfessionalが必要です。
こちらもインストール方法は世の中の優れた記事にお任せします。

ライブラリまわりについて

Javaをeclipseとかで開発するときのプロジェクトに相当するのはPythonでは仮想環境だと思います。
従って、まずは仮想環境を作ります。私の環境ではPyCharmを立ち上げたままAnacondaを入れると、再起動しないとcondaにパスが通ってない状態でした。

conda create -n my_env python numpy scipy

my_envが仮想環境の名前になります。python numpy scipyなどは最初からinstallしておきたいパッケージを指定します。
scipy=0.12.0のようにすることでバージョン指定も可能です。

仮想環境の一覧と仮想環境の切り替えは以下です。

conda info -e # 利用できる仮想環境の一覧を表示
activate my_env # Windows
source activate my_env # Mac/Linux

基本的に「conda insall hogepackage」でinstallしますが、見つからない場合があるので、良く見るパッケージ管理ツールpipもインストールします。

conda install pip

awsまわりのライブラリはboto3を利用します

pip install boto3

利用ライブラリの書き出しは以下です

pip freeze > requirements.txt

lambda-uploaderの利用

Lambdaへのアップロードにはlambda-uploaderを使います。

pip install lambda-uploader

50MB以下なら直接アップロードして適用できます。それ以上はlambda-uploaderでzipにして後は自分でS3にアップして適用するいつものやつになります。

AWSの認証情報との紐づけはprofileを作り、アップロード時に指定します。
とりあえずaws-cliが必要なので

pip install awscli

hoge-profileという名前でプロファイルを作成

aws configure --profile hoge-profile

AWS Access Key ID [None]: {Your Access Key Id}
AWS Secret Access Key [None]: {Your Secret Access Key}
Default region name [None]: ap-northeast-1
Default output format [None]: json

アップロード時のコマンド。

del *.zip
lambda-uploader --virtualenv=C:\hoge\my_env --profile=hoge-profile

コマンドの実行はlambda-uploaderの設定ファイルであるlambda.jsonが置かれているディレクトリで打つ前提です。
ビルドにミスるとゴミzipが出来上がって、次に実行する時にこのzipが含まれて大きなzipになるのでdelで削除しています。Linuxの場合は読み替えてください。
lambda-uploaderを使うとlambda.jsonとこのコマンドだけでアップロードが完了します。凄く便利。

lambda.jsonについて

lambda-uploaderの設定ファイルのlambda.jsonの内容は見れば分かるものばかりですが、注意すべきなのはname、handler、roleの３つだと思います。

{
  "name": "hogeFunc",
  "description": "write description",
  "region": "ap-northeast-1",
  "handler": "hoge.lambda_handler",
  "role": "arn:aws:iam::xxxxxxxxxxxx:role/hoge_lambda_exec_role",
  "timeout": 300,
  "memory": 128
}

nameはアップロード先のLambda Functionの名前と一致させる必要があります。
handlerはhoge.pyというファイルにlambda_handlerというメソッドを定義している場合です。Lambdaで実行されるメソッドになります。
roleはhogeFunc実行用のroleの値を入れる必要があります。

API GatewayのLambdaプロキシ統合を使う場合

API Gatewayの設定でLambdaプロキシ統合(Lambda Proxy Integration)を使うと普通に文字列をreturnするとinternal server errorで怒られます。以下のようにstatusCode body が必須、必要な情報はbodyに入れます。 'Access-Control-Allow-Origin': '*'を設定したheadersも入れることになると思います。

return { 'statusCode': 200, 'body': 'Hello from Lambda' }

Javaぽく書きたい！

Python3.5からType Hintsという機能が導入されました。
実行時には完全にコメント扱いなので好きなだけ書いていいです。
変数宣言の型と関数の戻り値の型などを明示できます。
Type Hintsの静的なチェックにはmypyを使います。

pip install mypy
mypy hoge.py

以下は私がLambdaのレスポンスをクラス化したいと思い、Pythonの命名規則の規約をガン無視してJavaぽく書いたコードになります。
自分で定義したclassに対してはjson.dumpsが使えないので、EntityBaseクラスにtoJson()メソッドをもたせ、継承してゴニョゴニョしています。
Lambdaで実行するメソッドの戻り値はDict[str, Any]でとりあえずは大丈夫でした。

def lambda_handler(event:Dict[str, Any], context) -> Dict[str, Any]:
    resultResponse:ResultResponse = ResultResponse(...)
    # 省略
    return resultResponse.toDict()

class ResultResponseBody(EntityBase, Generic[T]):
    def __init__(self:'ResultResponseBody', code:int, msg:str, data: object)  -> None:
        self.code:int = code
        self.msg:str = msg
        self.data:object = data

class ResultResponse(EntityBase, Generic[T]):
    def __init__(self:'ResultResponse', status:int, code:int, msg:str, data:object=None) -> None:
        # data:object={} とすると怒られるのでこの形が正しいらしい
        # https://docs.quantifiedcode.com/python-anti-patterns/correctness/mutable_default_value_as_argument.html
        if data is None:
            data = {}

        self.statusCode:int = status
        self.headers:Dict[str, str] = {  # Lambda_Proxyを有効にするのでCORSはAPI Gatewayではなくここで何とかする必要がある
            'Access-Control-Allow-Origin': '*'
        }
        #self.body:str = json.dumps(ResultResponseBody(code, msg, data)) # 自作クラスに対しては無力な模様
        self.body: str = ResultResponseBody(code, msg, data).toJson()

class EntityBase(Generic[T]):
    # from collections import namedtuple を使う方法もあるけど、、、
    def jsonToEntity(self:T, jsonString:str) -> T:
        dictObj = json.loads(jsonString)  # ファイルオブジェクトからのときはload 文字列からのときはloads
        for key in dictObj.keys():
            setattr(self, key, dictObj[key])
        return self

    # object的な感じで来たものを対象クラスに入れていく、keyはObjectの方をベースにする
    def translateTo(self:T, obj:Dict[str, Any]) -> T:
        for key in obj.keys():
            value:Any = obj[key]
            if value is None and isinstance(getattr(self, key), str):
                value = ''
            setattr(self, key, value)
        return self

    # なんか良い感じにjson.dumpsが動くかと思いきや、自作クラスは全く動かないという悲しい現実
    # relation等があって内部に別クラスを持つ場合はそのクラスがEntityBaseを継承していれば再帰的にうまくいくはず
    # それ以外の場合は。。。
    def toJson(self) -> str:
        res:Dict[str, Any] = {}
        for key in self.__dict__.keys():
            target:Any = getattr(self, key)

            if isinstance(target, EntityBase):
                res[key] = target.toJson()
                continue
            if isinstance(target, Decimal):
                decimalVal:Decimal = target
                strDecimalVal:str = str(decimalVal)
                if '.' not in strDecimalVal: 
                    res[key] = int(strDecimalVal)
                    continue
                else:
                    raise TypeError(key + strDecimalVal + " is invalid at base.py toJson")

            if isinstance(target, List):  # Listの中がentityの場合はここ
                listJson:str = ''
                for eachTarget in target:
                    if len(listJson) > 0:
                        listJson += ','
                    if isinstance(eachTarget, EntityBase):
                        listJson += eachTarget.toJson()
                    else:
                        listJson += json.dumps(eachTarget)
                res[key] = '[' + listJson + ']'
                continue

            if isinstance(target, Dict): # dataにEntityBaseを継承しないで直接Dictを突っ込んだ場合
                res[key] = json.dumps(target)
                continue

            res[key] = target

        return json.dumps(res)

    def toDict(self) -> Dict[str, Any]:
        ret:Dict[str, Any] = {}
        for key in self.__dict__.keys():
            ret[key] = getattr(self, key)
        return ret