PythonのデコレータでLambdaのコードをクリーンに保つ！！

サーバーレスアプリの開発では「フレームワークを使わない」という選択肢が採用されることも多いと思います。 フレームワークに頼らずにロジックを共通化し、Lambdaのコードをクリーンに保つ手法について調査したのでご紹介します。

岩田智哉

2018.10.29

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

サーバーレス開発部＠大阪の岩田です。 PythonにはDjangoやFlaskといったメジャーなアプリケーションフレームワークが存在しますが、サーバーレスアプリの開発においては「フレームワークを使用しない」という選択肢が採用されることも多いと思います。

フレームワークを使用しない場合、フレームワークがやってくれるような処理も全て自前で実装する必要があるため、自然とLambdaのコードが肥大化しがちです。 小規模な開発であれば、あまり問題になりませんが、中規模以上の開発になってくると、フレームワークを利用しないことの辛身が出てきます。

コードをクリーンに保つためのアプローチとしてPythonのデコレータ構文を試してみたのですが、うまく活用できそうな感触を得たので実装例をご紹介します。 なお、Pythonのバージョンは3.6を使用しています。

よくあるLambdaの実装

例えばデバイスのCRUD操作を行うREST APIをPythonとLambdaで実装するケースについて考えてみます。 APIの設計としては

  1. GET /devices でデバイス一覧取得
  2. POST /devices でデバイス登録
  3. GET /devices/{device_id} パスパラメータのdevice_idで指定されたデバイスの情報を取得
  4. PUT /devices/{device_id} パスパラメータのdevice_idで指定されたデバイスの情報を更新
  5. DELETE /devices/{device_id} パスパラメータのdevice_idで指定されたデバイスの情報を削除

のような設計が一般的かと思います。 一旦1.のデバイス一覧取得は無視して、各APIのざっくりした実装を見ていきたいと思います。

デバイス登録

#・・・・・省略
logger = logging.getLogger()
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Devices")

def lambda_handler(event, context):
 
    try:
        #・・・・・省略
        # リクエストボディに妥当なJSONが設定されているかチェック
        if not is_valid_json(body):
            return {
                "statusCode": HTTPStatus.BAD_REQUEST,
                "body": "Some error message",
            }
             
        # リクエストボディのJSONが妥当なスキーマかチェック
        if not is_valid_schema(body):
            return {
                "statusCode": HTTPStatus.BAD_REQUEST,
                "body": "Some error message",
        }       
 
        # API固有のバリデーション処理
        #・・・・・省略        
        # バリデーションOKならデバイスIDを採番して登録
        table.put_item(
            Item={
                "device_id": device_id,
                "foo": "bar"
            })
            
        #　登録できたらレスポンスを生成して返す
        #・・・・・省略        
        return {
            "statusCode": HTTPStatus.CREATED,
            "body": "some json",
        }
 
    except Exception as e:
        logger.error(e)
        raise e

※is_valid_jsonとis_valid_schemaは別途作成した自作関数です。

デバイス取得

#・・・・・省略
logger = logging.getLogger()
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Devices")

def lambda_handler(event, context):

    try:
        #・・・・・省略
        # パスパラメータからデバイスIDを取得してDynamoDBから検索        
        res = table.get_item(
            Key={
            "device_id": device_id,
        })
        
        # アイテムが存在しない場合は404を返す
        if "Item" not in res:
            return {
                "statusCode": HTTPStatus.NOT_FOUND,
                "body": "Not Found",
            }
        
        #　アイテムが存在する場合はレスポンスを生成して返す
        #・・・・・省略        
        return {
            "statusCode": HTTPStatus.OK,
            "body": "some json",
        }

    except Exception as e:
        logger.error(e)
        raise e

デバイス更新

#・・・・・省略
logger = logging.getLogger()
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Devices")

def lambda_handler(event, context):

    try:
        #・・・・・省略
        # パスパラメータからデバイスIDを取得してDynamoDBから検索        
        res = table.get_item(
            Key={
            "device_id": device_id,
        })
        
        # アイテムが存在しない場合は404を返す
        if "Item" not in res:
            return {
                "statusCode": HTTPStatus.NOT_FOUND,
                "body": "Not Found",
            }

        # リクエストボディに妥当なJSONが設定されているかチェック
        if not is_valid_json(body):
            return {
                "statusCode": HTTPStatus.BAD_REQUEST,
                "body": "Some error message",
            }
             
        # リクエストボディのJSONが妥当なスキーマかチェック
        if not is_valid_schema(body):
            return {
                "statusCode": HTTPStatus.BAD_REQUEST,
                "body": "Some error message",
        }       
 
        # API固有のバリデーション処理
        #・・・・・省略        
        # バリデーションOKなら更新
        table.put_item(
            Item={
                "device_id": device_id,
                "foo": "更新後の値"
            })
        
        #　アイテムが存在する場合は更新を行い、レスポンスを生成して返す
        #・・・・・省略        
        return {
            "statusCode": HTTPStatus.OK,
            "body": "some json",
        }

    except Exception as e:
        logger.error(e)
        raise e

※is_valid_jsonとis_valid_schemaは別途作成した自作関数です。

デバイス削除

#・・・・・省略
logger = logging.getLogger()
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Devices")

def lambda_handler(event, context):

    try:
        #・・・・・省略
        # パスパラメータからデバイスIDを取得してDynamoDBから検索        
        res = table.get_item(
            Key={
            "device_id": device_id,
        })
        
        # アイテムが存在しない場合は404を返す
        if "Item" not in res:
            return {
                "statusCode": HTTPStatus.NOT_FOUND,
                "body": "Not Found",
            }
        
        #　アイテムが存在する場合は削除してレスポンスを返す
        table.delete_item(
            Key={
                "device_id": device_id,
            })
        return {
            "statusCode": HTTPStatus.NO_CONTENT,
            "body": "some json",
        }

    except Exception as e:
        logger.error(e)
        raise e

ハイライト箇所はコードが冗長になっている箇所です。いくつかのAPIで

  • try〜exceptによる例外のハンドリング
  • 送信されたBody部が妥当なJSONかチェックし、NGの場合は400エラーを返す
  • パスパラメータのdevice_idをキーにデバイスの存在チェックを行い、見つからない場合は404エラーを返す

といった処理が重複しており、コードが冗長です。

また、こういった入力値のチェックやエラーハンドリングは品質の高いシステムを作るために重要なことではありますが、ビジネスロジックとは無関係で、開発者が実装したい本質的な問題ではありません。 本質的でないコードはもっと簡潔に記述し、コードベースをスッキリした状態に保ちたいところです。

共通処理をデコレータに切り出してみる。

冗長な処理をデコレータに切り出していきます。下記のような実装にしてみました。

handle_exception

def handle_exception(logger):
    def _handle_exception(func):
        def handle_exception_wrapper(*args, **kwargs):

            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.error(e)
                raise e
        return handle_exception_wrapper

    return _handle_exception

例外のハンドリングを行うためのデコレータです。 元の関数にtry〜exceptブロックを追加し、例外をキャッチした際はログを出力します。 try〜exceptブロックを追加するという特性から、このデコレータは他のデコレータよりも先に適用すべきです。 関数をデコレートする部分の一番上に持ってくるようにしましょう。

valid_json_check

def valid_json_check(func):
    def is_valid_json(body):
        # JSON形式として妥当かをチェックする処理
        
    def json_check_wrapper(*args, **kwargs):

        body = args[0]["body"]
        if not is_valid_json(body):
            return {
                "statusCode": HTTPStatus.BAD_REQUEST,
                "body": "Some error message",
            }
        return func(*args, **kwargs)

    return json_check_wrapper

送信されてきたBody部に妥当なJSONが含まれるかチェックするデコレータです。 例えば{"key":"val}のような値が送信されてきた場合は400エラーを返します。

valid_json_schema_check

def valid_json_schema_check(schema):
    def _valid_json_schema_check(func):
        def is_valid_json_schema(schema):
            # Cerberusを使ってJSONのスキーマをチェックする処理
        def json_check_wrapper(*args, **kwargs):
    
            body = args[0]["body"]
            if not is_valid_json_schema(schema):
                return {
                    "statusCode": HTTPStatus.BAD_REQUEST,
                    "body": "Some error message",
                }    
            return func(*args, **kwargs)    
            
        return json_check_wrapper
        
    return _valid_json_schema_check

送信されてきたJSONが妥当なスキーマかチェックするデコレータです。 実際のチェック処理はCerberusというライブラリを利用しており、デコレータの引数に渡された辞書オブジェクトを元に諸々のチェックを行います。

device_exists_check

def device_exists_check(dynamo_table):
    def _device_exists(func):
        def device_exists_check_wrapper(*args, **kwargs):

            device_id = args[0]['pathParameters']['device_id']
            res = dynamo_table.get_item(
                Key={
                "device_id": device_id,
            })

            if not 'Item' in res:
                return {
                    "statusCode": HTTPStatus.NOT_FOUND,
                    "body": "Not Found",
                }
            return func(*args, res['Item'], **kwargs)

        return device_exists_check_wrapper

    return _device_exists

パスパラメータのdevice_idを元にデバイスの存在チェックを行うデコレータです。 デバイスが存在しない場合は404エラーを返却します。 デバイスが存在した場合、後続のロジックで対象デバイスの情報を利用できるようにデコレート元の関数を呼び出す際の引数にデバイスの情報を追加しています。 ※ハイライト箇所

デコレータを使ってリファクタリングしてみる

作成したデコレータを使って、コードをリファクタリングしてみました。

デバイス取得

#・・・・・省略
logger = logging.getLogger()
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Devices")

@handle_exception(logger)
@valid_json_check
@device_exists_check(table)
def lambda_handler(event, context, device):
 
    #　アイテムが存在する場合はレスポンスを生成して返す      
    return {
        "statusCode": HTTPStatus.OK,
        "body": "some json",
    }

device_exists_checkでデコレートしているので、ハンドラの引数にデバイスの情報を保持するdeviceという変数を追加しています。 デコレータが上から下に順番に適用されていくことを意識して、適切な順番でデコレートしていきます。

デバイス登録

#・・・・・省略
schema = {
    "foo": {
        'type': 'string',
        'required': True,
    }
@handle_exception(logger)
@valid_json_check
@valid_json_schema_check(schema)
def lambda_handler(event, context): 

    # API固有のバリデーション処理
    #・・・・・省略        
    # バリデーションOKならデバイスIDを採番して登録
    table.put_item(
        Item={
            "device_id": device_id,
            "foo": "bar"
        })
         
    #　登録できたらレスポンスを生成して返す
    #・・・・・省略        
    return {
        "statusCode": HTTPStatus.CREATED,
        "body": "some json",
    }

こちらはあまり特筆すべきことはありません。

デバイス更新

#・・・・・省略    
schema = {
    "foo": {
        'type': 'string',
        'required': True,
    }
@handle_exception(logger)
@valid_json_check
@valid_json_schema_check(schema)
@device_exists_check(table)
def lambda_handler(event, context, device):
 
    # API固有のバリデーション処理
    #・・・・・省略
    # バリデーションOKなら更新
    table.put_item(
        Item={
            "device_id": device['device_id'],
            "foo": "更新後の値"
        })
         
    # レスポンスを生成して返す
    #・・・・・省略        
    return {
        "statusCode": HTTPStatus.OK,
        "body": "some json",
    }

デバイス取得と同様ハンドラの引数にデバイスの情報を保持するdeviceという変数を追加しています。

デバイス削除

#・・・・・省略
logger = logging.getLogger()
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Devices")

@handle_exception(logger)
@valid_json_check
@device_exists_check(table)
def lambda_handler(event, context, device):
 
    table.delete_item(
        Key={
            "device_id": device['device_id'],
        })   
    return {
        "statusCode": HTTPStatus.NO_CONTENT,
        "body": "some json",
    }

デバイス取得と同様ハンドラの引数にデバイスの情報を保持するdeviceという変数を追加しています。

元のコードに比べると、非常にスッキリしたコードになりました！！ 決まり切ったエラーチェック等を排除できたので、メインとなるビジネスロジックの開発に集中できそうです。

まとめ

Pythonのデコレータ構文を使った処理の共通化について見て来ました。 処理を共通化するための方法としてはクラスの汎化・特化を使う方法など様々なアプローチが考えられますが、気軽に処理を付け外しできるのは、デコレータの大きなメリットだと感じました。 クラス設計が悩ましい場合等は、まずはお気軽にデコレータでリファクタリングしていくというアプローチも有効なのではないでしょうか？ 誰かのお役に立てば幸いです。

AWS

関連記事

AWS Direct Connect のフェイルオーバーテストを使って待機系の AWS Site-to-Site VPN へ切り替えテストをしてみた

いわさ

2024.04.20

AWS入門ブログリレー2024〜AWS CloudFormation編〜

平木佳介

2024.04.19

Amazon IVSリアルタイムストリーミング機能｜視聴までの処理を見てみよう！

坂齊史奈子

2024.04.19

Amplify Gen2のDataのモデル定義が分かりやすい

中村佳央

2024.04.19