こんにちは。ゴーリスト開発の盛次です。
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
まとめ
アップロード周りはlambda-uploaderの使い勝手が非常に良いです。
Type Hintsはかなり良いと思います。Pythonでも型に基づいてIDEのサポートをフルに受けたいという人にはお勧めです。
