kakakakakku blog

Weekly Tech Blog: Keep on Learning!

Terraform で無料利用枠の VPC IP Address Manager (IPAM) を設定する

2023年11月から VPC IP Address Manager (IPAM)「無料枠利用枠」が追加されて Public IP Insights などの機能が無料で使えるようになった💡そして,2024年2月から課金対象になった IPv4 の最適化のために Public IP Insights を使いたいという場面もあると思う.

aws.amazon.com

aws.amazon.com

Terraform で試す

実は Terraform AWS Provider では今まで aws_vpc_ipamtier はサポートされていなかった.今日(2024年3月29日)にリリースされた v5.43.0 でついにサポートされた❗️待ってました〜 \( 'ω')/

github.com

👾 ipam.tf

設定自体は簡単で aws_vpc_ipamtier = "free" を追加すれば OK👌デフォルトは advanced なので注意しておくと良さそう.あと今回は operating_regions にバージニア北部リージョンを設定した💡もちろん複数リージョンを設定することもできる.

resource "aws_vpc_ipam" "main" {
  tier = "free"
  operating_regions {
    region_name = "us-east-1"
  }
}

関連記事

AWS CDK で VPC IP Address Manager (IPAM) を設定する記事は過去に書いているので参考まで〜📝

kakakakakku.hatenablog.com

AWS CDK で外部パッケージを含む Python の AWS Lambda 関数をデプロイする

AWS CDK で外部パッケージを含む Python の AWS Lambda 関数をデプロイする場合,requirements.txt から依存関係を解決して,デプロイするアセットとして ZIP にまとめる(バンドルする)必要がある💡

今回は aws-cdk-lib.aws_lambda module@aws-cdk/aws-lambda-python-alpha module を使う方法を試す❗️

前提

今回はサンプルとして requests に依存したコードを以下のディレクトリ構成で置いてある前提とする \( 'ω')/

functions/requests
├── app.py
└── requirements.txt

aws_lambda module を使う

まず,AWS CDK で AWS Lambda 関数をデプロイするときによく使う aws_lambda module では Code.fromAsset のオプションとして BundlingOptions を設定できる.

docs.aws.amazon.com

実装としてはザッとこんな感じになる 👾

Python のコンテナ内で pip install コマンドを実行して /asset-output ディレクトリに依存関係をインストールしたら,後はそのまま ZIP にまとめて(バンドルして)デプロイされる👌

new aws_lambda.Function(this, 'PythonFunction', {
  functionName: 'sandbox-python-function',
  runtime: aws_lambda.Runtime.PYTHON_3_12,
  handler: 'app.lambda_handler',
  code: aws_lambda.Code.fromAsset(path.join(__dirname, '../functions/requests'), {
    bundling: {
      image: aws_lambda.Runtime.PYTHON_3_12.bundlingImage,
      command: [
        'bash', '-c',
        'pip install -r requirements.txt -t /asset-output && cp -au . /asset-output'
      ]
    }
  })
})

aws-lambda-python-alpha module を使う

まだ alpha ではあるけど aws-lambda-python-alpha module を使うと entry に指定したディレクトリにある requirements.txtPipfile から自動的に依存関係を解決してくれる👌

実装としてはザッとこんな感じになる 👾

new aws_lambda_python_alpha.PythonFunction(this, 'PythonFunctionAlpha', {
  functionName: 'sandbox-python-function-alpha',
  runtime: aws_lambda.Runtime.PYTHON_3_12,
  index: 'app.py',
  handler: 'lambda_handler',
  entry: path.join(__dirname, '../functions/requests'),
})

aws_lambda.Function のプロパティもサポートされてるし,Lambda Layer も簡単にデプロイできるし便利〜 \( 'ω')/

docs.aws.amazon.com

デプロイ確認

期待通りに requirements.txt に定義した requests をデプロイできている❗️

JSON Schema で簡単にバリデーションを実装できる Powertools for AWS Lambda (Python) の Validation

Powertools for AWS Lambda (Python)「Validation」を使うと AWS Lambda 関数に渡されたイベント情報のバリデーションを JSON Schema に沿って実現できる.例えば,必須パラメータ・文字数制限・ENUM・正規表現などをチェックできる👌

Powertools for AWS Lambda (Python) 自体は Tracer / Logger / Event Source Data Classes などをよく使うけど,Validation は今まで活用できてなく,試してみたらとても便利だったので,今回試した結果を簡単にまとめておく \( 'ω')/

docs.powertools.aws.dev

検証環境

今回は AWS SAM を使って Amazon API Gateway (REST API) と AWS Lambda 関数を構築する.あくまでサンプルとして Amazon API Gateway の / に POST リクエストを送るとバリデーションロジックを含んだ AWS Lambda 関数が実行されるようにした.また Powertools は Lambda Layer でセットアップする.

AWSTemplateFormatVersion: 2010-09-09
Transform: AWS::Serverless-2016-10-31

Resources:
  Function:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: powertools-validation
      CodeUri: src/
      Handler: app.lambda_handler
      Runtime: python3.12
      Architectures:
        - x86_64
      Layers:
        - arn:aws:lambda:ap-northeast-1:017000801446:layer:AWSLambdaPowertoolsPythonV2:67
      Events:
        Api:
          Type: Api
          Properties:
            Path: /
            Method: POST

最終的なディレクトリ構成は以下のようになる💡

├── events
│   └── event.json
├── samconfig.toml
├── src
│   └── app.py
└── template.yaml

👾 app.py

今回はバリデーションを紹介するサンプルとして,特にロジックはなく Amazon API Gateway への POST リクエストに対して 200 OK もしくは 400 BAD_REQUEST を返す実装にした.また AWS Lambda 関数のベストプラクティスを参考に Handler とロジックを分割して main() にまとめてある(実際にはもっと細かく分割しても良さそう).

そして,今回は超簡易的な「TODO アプリ」を例として,title / category / description / link というパラメータを受け取る API のバリデーションを実装した.パラメータごとのバリデーション要件は以下のコードの SCHEMA を見てもらえればと❗️さらに Powertools for AWS Lambda (Python) の Validation ではバリデーションロジックを @validator デコレータを使った実装と validate() 関数を使った実装から選べる.どちらも試してみて,個人的には以下の2つの理由から validate() 関数を使うのが良いと思った💡

  • validate() 関数は Lambda コンテキストに依存してなくてローカル開発がしやすかった
  • 例外発生時のハンドリングなどを柔軟に実装しやすかった

また validate() 関数を呼び出すときに envelope を指定できて,イベントオブジェクトの中からバリデーションする箇所を限定できる.今回は Amazon API Gateway (REST API) から渡されるイベントをバリデーションするため,envelopes.API_GATEWAY_REST を指定した.すると自動的に body がバリデーション対象になる👌現状は8種類の envelope が提供されている \( 'ω')/

  • API_GATEWAY_HTTP
  • API_GATEWAY_REST
  • CLOUDWATCH_EVENTS_SCHEDULED
  • CLOUDWATCH_LOGS
  • EVENTBRIDGE
  • KINESIS_DATA_STREAM
  • SNS
  • SQS
import json
from aws_lambda_powertools.utilities.validation import SchemaValidationError, envelopes, validate
from http import HTTPStatus
from jmespath.exceptions import JMESPathTypeError

SCHEMA = {
    '$schema': 'http://json-schema.org/draft-07/schema',
    'type': 'object',
    'required': ['title', 'category'],
    'properties': {
        'title': {
            'type': 'string',
            'pattern': '^[a-zA-Z0-9_]*$'
        },
        'category': {
            'type': 'string',
            'enum': ['Python', 'Go', 'Java']
        },
        'description': {
            'type': 'string',
            'minLength': 10,
            'maxLength': 100
        },
        'link': {
            'type': 'string',
            'format': 'uri'
        }
    },
}


def main(event):
    try:
        validate(event=event, schema=SCHEMA, envelope=envelopes.API_GATEWAY_REST)
    except JMESPathTypeError as e:
        return {
            'statusCode': HTTPStatus.BAD_REQUEST,
            'body': json.dumps({'message': str(e)})
        }
    except json.JSONDecodeError as e:
        return {
            'statusCode': HTTPStatus.BAD_REQUEST,
            'body': json.dumps({'message': e.msg})
        }
    except SchemaValidationError as e:
        return {
            'statusCode': HTTPStatus.BAD_REQUEST,
            'body': json.dumps({'message': e.validation_message})
        }

    return {
        'statusCode': HTTPStatus.OK,
        'body': json.dumps({'message': 'ok'})
    }


def lambda_handler(event, context):
    return main(event)


if __name__ == '__main__':
    with open('../events/event.json', 'r') as f:
        event = json.load(f)

    main(event)

動作確認

data must contain ['category', 'title'] properties

event.json

{}

必須の titlecategory プロパティがなくバリデーションエラー🙅‍♂️

$ curl -s -X POST --data @event.json ${ENDPOINT}
{"message": "data must contain ['category', 'title'] properties"}

data.title must match pattern ^[a-zA-Z0-9_]*$

event.json

{
    "title": "Powertoolsを試す",
    "category": "Python"
}

title に日本語を含んでいるためバリデーションエラー🙅‍♂️

$ curl -s -X POST --data @event.json ${ENDPOINT}
{"message": "data.title must match pattern ^[a-zA-Z0-9_]*$"}

data.category must be one of ['Python', 'Go', 'Java']

event.json

{
    "title": "Powertools",
    "category": "AWS"
}

category の ENUM 以外の値を指定しているためバリデーションエラー🙅‍♂️

$ curl -s -X POST --data @event.json ${ENDPOINT}
{"message": "data.category must be one of ['Python', 'Go', 'Java']"}

data.description must be longer than or equal to 10 characters

event.json

{
    "title": "Powertools",
    "category": "Python",
    "description": ""
}

description が10文字以上になっていないためバリデーションエラー🙅‍♂️

$ curl -s -X POST --data @event.json ${ENDPOINT}
{"message": "data.description must be longer than or equal to 10 characters"}

event.json

{
    "title": "Powertools",
    "category": "Python",
    "description": "Try Powertools for AWS Lambda.",
    "link": "docs.powertools.aws.dev"
}

link が URL 形式になっていないためバリデーションエラー🙅‍♂️

$ curl -s -X POST --data @event.json ${ENDPOINT}
{"message": "data.link must be uri"}

まとめ

Powertools for AWS Lambda (Python)「Validation」を使うと AWS Lambda 関数に渡されたイベント情報のバリデーションを柔軟に実装できてとても便利だった❗️

今後は積極的に使っていくぞー \( 'ω')/

関連リンク🔗

github.com

testcontainers-python: pytest 実行時に使い捨て可能な LocalStack を起動する

Testcontainers を使うと,テストコードを実行するときに必要になるデータベース・キャッシュ・キューなどの依存関係をコード上で管理できて,実行後にはコンテナを自動的に消してくれるという使い捨て可能な仕組みを簡単に作れる❗️Testcontainers のサイトに載っている「Test dependencies as code」という表現はピッタリだと思う👌

testcontainers.com

Testcontainers は Java / Go / .NET / Rust など多くの言語をサポートしているけど,今回は Python 用の testcontainers-python を試してみた.検証に使ったコードを紹介しつつ,簡単にまとめておく✍

また Testcontainers Cloud もあったりする🌩

testcontainers.com

前提

今回は以下の前提で試す💡なお testcontainers-python は MySQL / PostgreSQL / Redis / Kafka / RabbitMQ / LocalStack など多くの依存関係をサポートしているので,組み合わせてテストコードを実行することもできる.今回は LocalStack に限定する.

  • Amazon DynamoDB を操作するコードをテストする
  • Amazon DynamoDB のエミュレーターとして LocalStack を使う
  • テストフレームワークとして pytest を使う
  • testcontainers-pythonLocalStackContainer を使う

ディレクトリ構成は以下のようにした.特に決まってなく自由に変更できる👌

.
├── README.md
├── pyproject.toml
├── requirements-test.txt
├── src
│   └── app.py
├── tests
│   └── test_app.py
└── venv

ドキュメント

testcontainers-python に関しては以下のドキュメントを読むとイメージがつかめると思う.

testcontainers-python.readthedocs.io

testcontainers.com

しかし,残念ながら testcontainers-python の LocalStackContainer に関するドキュメントはほとんどなく(見つけられず)実際に GitHub のコードを読みながらメソッドなどを探したりしていた.

github.com

サンプルコード

👾 app.py

コードには特に意味はないけど,Amazon DynamoDB の Forum テーブルからアイテムを取得する search_forum() 関数を今回のテスト対象とする.Forum テーブルというのは Amazon DynamoDB のドキュメントに載っているサンプルでそのまま使うことにした.

そして,boto3 client は環境変数 ENV によって3種類作れるようにしてある👌

  • local: ローカル開発用 (LocalStack)
  • test: テスト用 (testcontainers-python / LocalStack)
  • その他: 実際の AWS アカウント

また testcontainers-python の LocalStackContainer を使うと http://localhost:63033http://localhost:63058 など実行時のポートが変化するため,後述する test_app.py で環境変数 TESTCONTAINERS_LOCALSTACK_ENDPOINT_URL を設定することにした👌もっとイイ方法もありそう.ちなみに LocalStackContaineredge_port0.0.0.0:63167->24566/tcp のようにネットワークが構成されるため "コンテナ側のポート" を指定するオプションだった.

import boto3
import os

TABLE_NAME = 'Forum'


if os.environ['ENV'] == 'local':
    dynamodb = boto3.client('dynamodb', endpoint_url='http://localhost:4566')
elif os.environ['ENV'] == 'test':
    dynamodb = boto3.client('dynamodb', endpoint_url=os.environ['TESTCONTAINERS_LOCALSTACK_ENDPOINT_URL'])
else:
    dynamodb = boto3.client('dynamodb')


def search_forum(name):
    return dynamodb.get_item(
        TableName=TABLE_NAME,
        Key={'Name': {'S': name}}
    )

👾 test_app.py

pytest 実行時に呼び出すフィクスチャとして @pytest.fixture デコレータで setup() 関数を実装した.LocalStackContainer で LocalStack のコンテナイメージを設定しているため,テストコードを実行する前に自動的に LocalStack コンテナが起動される👌ちなみに LocalStackContainerget_client() 関数は boto3 client を返しているため,boto3 に慣れていれば普段と同じように実装できる.

そして,Amazon DynamoDB テーブル Forum を作って,サンプルデータ(アイテム)を1つ登録している.テスト観点によっては Faker などを使ってリアルなサンプルデータを登録すると良さそう.

最後にテストケースとしては search_forum() 関数を呼び出して「アイテムを取得できる場合」「アイテムを取得できない場合」を確認している✔️

import os
import pytest
from testcontainers.localstack import LocalStackContainer

TABLE_NAME = 'Forum'


@pytest.fixture(scope='module', autouse=True)
def setup():
    with LocalStackContainer(image='localstack/localstack:3', region_name='ap-northeast-1') as localstack:
        os.environ['TESTCONTAINERS_LOCALSTACK_ENDPOINT_URL'] = localstack.get_url()

        dynamodb = localstack.get_client('dynamodb')

        dynamodb.create_table(
            TableName=TABLE_NAME,
            KeySchema=[
                {
                    'AttributeName': 'Name',
                    'KeyType': 'HASH',
                }
            ],
            AttributeDefinitions=[
                {
                    'AttributeName': 'Name',
                    'AttributeType': 'S',
                }
            ],
            BillingMode='PAY_PER_REQUEST',
        )

        item = {
            'Name': {'S': 'Amazon DynamoDB'},
            'Category': {'S': 'Amazon Web Services'},
            'Threads': {'N': '2'},
            'Messages': {'N': '4'},
            'Views': {'N': '1000'},
        }

        dynamodb.put_item(TableName=TABLE_NAME, Item=item)

        yield localstack


def test_search_forum():
    from app import search_forum
    
    item = search_forum('Amazon DynamoDB')
    assert item['Item']['Category']['S'] == 'Amazon Web Services'
    assert item['Item']['Views']['N'] == '1000'

    item = search_forum('Amazon S3')
    assert 'Item' not in item

testcontainers-python と LocalStack の検証はザッとこんな感じ \( 'ω')/

動作確認

期待通り実行できた👏

$ ENV=test pytest -p no:warnings
=========================================================================================== test session starts ============================================================================================
(中略)
configfile: pyproject.toml
collected 1 item

tests/test_app.py .                                                                                                                                                                                  [100%]

============================================================================================ 1 passed in 4.97s =============================================================================================

現時点だと Python 3.12 で boto3 の DeprecationWarning が出るため -p no:warnings で抑止している🛑

github.com

Lambda オーソライザーのポリシーを簡単に出力できる Powertools for AWS Lambda (Python) の APIGatewayAuthorizerResponse

Amazon API Gateway の Lambda オーソライザー(旧カスタムオーソライザー)を使ってアクセス制御をするときに Lambda オーソライザーの仕様に沿ったポリシーを出力する必要がある💡詳しくは以下のドキュメントに載っている.

docs.aws.amazon.com

今まではドキュメントに載っているコードを参考に実装することが多かったけど,Powertools for AWS Lambda (Python)Event Source Data ClassesAPIGatewayAuthorizerResponse を使うと比較的簡単にポリシーを出力できて良かった❗️

docs.powertools.aws.dev

サンプルコード

今回トークンベースの Lambda オーソライザーで検証したときに使ったコードを載せておく📝

from aws_lambda_powertools.utilities.data_classes import event_source
from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (APIGatewayAuthorizerTokenEvent, APIGatewayAuthorizerResponse)
from aws_lambda_powertools.utilities.typing import LambdaContext


@event_source(data_class=APIGatewayAuthorizerTokenEvent)
def lambda_handler(event: APIGatewayAuthorizerTokenEvent, context: LambdaContext):
    arn = event.parsed_arn

    policy = APIGatewayAuthorizerResponse(
        principal_id='user',
        region=arn.region,
        aws_account_id=arn.aws_account_id,
        api_id=arn.api_id,
        stage=arn.stage
    )

    if event.authorization_token == 'allow':
        policy.context = {
            'key1': 'value1',
            'key2': 'value2',
            'key3': 'value3'
        }
        policy.allow_all_routes()
    else:
        policy.deny_all_routes()

    return policy.asdict()

ドキュメントに載っている Authorization ヘッダーに allow という文字列が設定されていれば OK という簡易的な実装だけど,ポリシーを出力するときは APIGatewayAuthorizerResponse に値を設定して allow_all_routes() 関数もしくは deny_all_routes() 関数で出力すれば良くて簡単👍

また認証後に実行される AWS Lambda 関数などに追加情報を渡す場合は context に dict で Key-Value を設定すれば OK👌

docs.aws.amazon.com

動作確認

$ curl -H 'Authorization: allow' ${ENDPOINT}
ok

$ curl -H 'Authorization: deny' ${ENDPOINT}
{"Message":"User is not authorized to access this resource with an explicit deny"}