Pipenv+LocalStackで作るLambda開発環境

from model import Handler
import boto3
import os

if os.getenv("LOCALSTACK_HOSTNAME") is None:
    s3 = boto3.client("s3", "ap-northeast-1")
else:
    endpoint = f"http://{os.environ['LOCALSTACK_HOSTNAME']}:4566"
    s3 = boto3.client(
        service_name="s3",
        endpoint_url=endpoint,
        aws_access_key_id="test",
        aws_secret_access_key="test",
    )


def lambda_handler(event, context) -> str:
    handler = Handler(event, context, s3)
    return handler.main()

import logging
import pandas as pd
import tempfile
import json
from typing import List


logger = logging.getLogger()
logger.setLevel(logging.INFO)


class Handler(object):
    def __init__(self, event, context, s3):
        self.event = event
        self.context = context
        self.s3 = s3

    def main(self) -> str:
        try:
            bucket = "test-bucket"
            send = "test.xlsx"
            data_path = self.event["input_obj"]
            dict_data: List[dict] = self.get_s3_data(bucket, data_path)
            df = self.make_df(dict_data)
            df_processed = self.process(df)
            send = self.send_excel(df_processed, bucket, send)
            return "completed : {0}".format(send)

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

    def get_s3_data(self, bucket: str, key: str) -> List[dict]:
        resp = self.s3.get_object(Bucket=bucket, Key=key)
        body = resp["Body"].read().decode("utf-8")
        json_dict: List[dict] = json.loads(body)
        return json_dict

    def make_df(self, data: list) -> pd.DataFrame:
        df = pd.DataFrame.from_dict(data)
        return df

    def calc(self, row):
        if row["会員ランク"] > 3:
            return row["ポイント"] * 1.25
        else:
            return row["ポイント"]

    def process(self, data: pd.DataFrame) -> pd.DataFrame:
        data["ボーナスポイント"] = data.apply(self.calc, axis=1)
        return data

    def send_excel(self, df: pd.DataFrame, bucket: str, send: str) -> str:
        with tempfile.TemporaryFile() as fp:
            writer = pd.ExcelWriter(fp, engine="xlsxwriter")
            df.to_excel(writer, sheet_name="Sheet1", index=False)
            writer.save()
            fp.seek(0)
            self.s3.put_object(
                Body=fp.read(),
                Bucket=bucket,
                Key=send,
            )
        return send

import pytest
from model import Handler
import pandas as pd


@pytest.fixture
def handler():
    return Handler(
        event={},
        context={},
        s3="",
    )


@pytest.mark.parametrize(
    "input_dict,expected_dict",
    [
        (
            [
                {
                    "会員番号": "000",
                    "名前": "椎名 米子",
                    "会員ランク": 1,
                    "ポイント": 45,
                    "タイムスタンプ": "2021-07-14",
                },
                {
                    "会員番号": "001",
                    "名前": "広島 たくみ",
                    "会員ランク": 4,
                    "ポイント": 39,
                    "タイムスタンプ": "2021-12-17",
                },
                {
                    "会員番号": "002",
                    "名前": "大嶺 順子",
                    "会員ランク": 2,
                    "ポイント": 27,
                    "タイムスタンプ": "2021-09-23",
                },
            ],
            [
                {
                    "会員番号": "000",
                    "名前": "椎名 米子",
                    "会員ランク": 1,
                    "ポイント": 45,
                    "タイムスタンプ": "2021-07-14",
                    "ボーナスポイント": 45,
                },
                {
                    "会員番号": "001",
                    "名前": "広島 たくみ",
                    "会員ランク": 4,
                    "ポイント": 39,
                    "タイムスタンプ": "2021-12-17",
                    "ボーナスポイント": 48.75,
                },
                {
                    "会員番号": "002",
                    "名前": "大嶺 順子",
                    "会員ランク": 2,
                    "ポイント": 27,
                    "タイムスタンプ": "2021-09-23",
                    "ボーナスポイント": 27,
                },
            ],
        ),
    ],
)
def test_process(handler, input_dict, expected_dict):
    json_dict = pd.DataFrame.from_dict(input_dict)
    got = handler.process(json_dict).sort_index(axis=1, ascending=False)
    expected = pd.DataFrame.from_dict(expected_dict).sort_index(axis=1, ascending=False)
    pd.testing.assert_frame_equal(got, expected)

import datetime
import json
from random import randint

import boto3
from fire import Fire
from mimesis import Person
from mimesis.locales import Locale

person = Person(Locale.JA)


def dummy_data(num: int) -> dict:
    id = str(num).zfill(3)
    date = datetime.date(2021, randint(1, 12), randint(1, 28)).strftime("%Y-%m-%d")
    dummy_dict = {
        "会員番号": id,
        "名前": person.full_name(reverse=True),
        "会員ランク": randint(1, 5),
        "ポイント": randint(50, 100),
        "タイムスタンプ": date,
    }
    return dummy_dict


def send_json(s3, sample_data: list, bucket: str, send: str) -> str:
    with open("utils/data/sample_data.json", mode="wt", encoding="utf-8") as f:
        json.dump(sample_data, f, ensure_ascii=False, indent=4)
    s3.put_object(
        Body=json.dumps(sample_data, ensure_ascii=False, indent=4),
        Bucket=bucket,
        Key=send,
    )
    return send


def make_dummy_data(k) -> list:
    sample_data = []
    for i in range(k):
        sample_data.append(dummy_data(i))
    return sample_data


def main(iterate_num: int) -> str:
    endpoint = f"http://localhost:4566"
    s3 = boto3.client(
        service_name="s3",
        endpoint_url=endpoint,
        aws_access_key_id="test",
        aws_secret_access_key="test",
    )
    bucket = "test-bucket"
    send = "test.json"
    sample_data = make_dummy_data(iterate_num)
    send = send_json(s3, sample_data, bucket, send)
    return send


if __name__ == "__main__":
    Fire(main)