祝日取得サーバーを作ってみよう！

どもども。
今回のお題の通り、祝日を管理してくれる簡単なサーバーを作っていきます。
主に活用するサービスや言語は以下、

• python
• docker
• aws

目的

私が携わっているプロジェクトで祝日を取得する必要が出たこと、当初はサービスに組み込もうと思いましたが、それだと他のサービスで必要となった場合、汎用性がないと判断したため祝日を管理することを各サービス持つのは良くないと判断したため、祝日を管理取得するサーバー（レス）を構築しました。

アーキテクチャ

全体の構成図です。GCPとAWSのクラウド上で完結しています。

こだわった点

なるべく、ローコスト、ローメンテナンスでもう2度と触らないようにした。

難しかった点

pythonでの処理はそこまで難しくありません。
どちらかというとGCPやAWSの知識が今回新たに勉強した点なので、そこが難しかったかなと。
ロールとポリシーへの理解が必要になってきます。

google calendar

祝日を管理する上で、祝日を正確にかつ未来永劫に取得できるであろうサイトとしてgoogle以上のものはないと判断したので、googleの祝日カレンダーから取得しています。
必要な情報は、APIのトークンと、取得したいカレンダーアドレス。

# 今回は日本の祝日のみ
ja.japanese#[e-mail]

aws

当初はEKSでの構築を考えていましたが、AWS Lambdaでもコンテナイメージが使えるとわかったので、そちらを活用しました。
サーバーレスではありますが、pipfileのバージョン管理はしなければならないので、完全に手が離れたわけではない・・・。
まぁ、最悪EKSにしようかなと思ってたりもする。

じゃ作ってみっか！

構成としては、

• ディレクトリ構成
• dockerfileの構成
• Pipfileの構成
• 祝日を取得しS3に保存する
• requestに対してS3のjsonを取得、加工してresponseする
• s3接続用のクライアントクラス

ディレクトリの構成

root
    /docker
        dockerfile
        docker-compose.yml
    /holidays
        app.py
        build_holiday.py
        get_holiday.py
    Pipfile
    Pipfile.lock

dockerfileの構成

マルチステージングビルドへ変更してもいい気がするが、まぁそこまで容量削れなかったのでまぁいいかという感じ。

# 無駄にpython3.12
FROM public.ecr.aws/lambda/python:3.12

COPY ./Pipfile /tmp/Pipfile
COPY ./Pipfile.lock /tmp/Pipfile.lock

RUN python3 -m pip install --upgrade pip && \
    python3 -m pip install pipenv && \
    PIPENV_PIPFILE=/tmp/Pipfile pipenv sync --system --dev
    
COPY holiday ${LAMBDA_TASK_ROOT}

CMD ["app.handler"]

Pipfileの構成

python3.12を使っているが、あまり使いこなせてない。内容としては3.11でもOK

[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
google-api-python-client = "==2.111.0"
google-auth-httplib2 = "==0.2.0"
google-auth-oauthlib = "==1.2.0"

[requires]
python_version = "3.12"

祝日を取得しS3に保存する

google APIカレンダーから祝日を取得してAWS S3にJSONとして保存する

build_holidays.py

import json
import time
from botocore.exceptions import ClientError
from collections import defaultdict
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta
from googleapiclient.discovery import build, Resource


# GOOGLE_API_KEYとS3は各自で準備してね
GOOGLE_API_KEY = "api_key"
CALENDAR_ID = "ja.japanese#[e-mail]"
S3_BUCKET = "holidays"

def dict_rec_update(d: dict, additional: dict) -> dict:
    """
    dictの再帰的なupdate
    """
    for k, v in additional.items():
        if isinstance(v, dict):
            additional[k] = dict_rec_update(d.get(k, {}), v)
    return d | additional


def ddict2dict(d: defaultdict | dict) -> dict:
    """
    defaultdictからdictに変換する
    """
    for k, v in d.items():
        if isinstance(v, dict):
            d[k] = ddict2dict(v)
    return dict(d)


def request_google_api(
    service: Resource,
    calendar_id: str,
    start_date: datetime,
    end_date: datetime
) -> dict:
    conditions = {
        'calendarId': calendar_id,
        'timeMin': start_date.isoformat() + 'Z',
        'timeMax': end_date.isoformat() + 'Z',
        'singleEvents': True,
        'orderBy': 'startTime'
    }
    return service.events().list(**conditions).execute()

def update_or_create_json(name: str, events: dict):
    s3 = boto3.client('s3')
    # s3上からの読み込み処理
    try:
        _json = json.loads(s3.download(f"{name}.json") or '{}')
    except ClientError:
        _json = {}

    if events:
        holidays = defaultdict(
            lambda: defaultdict(lambda: defaultdict(dict))
        )
        for event in events:
            _date = datetime.strptime(
                event['start'].get('date'),
                "%Y-%m-%d"
            ).date()
            summary = event['summary']
            holidays[f'{_date.year}'][f'{_date.month}'][
                f'{_date.day}'] = {
                    "summary": summary,
                    "date": _date.isoformat()
                }
        _json = dict_rec_update(_json, ddict2dict(holidays))

    # S3書き込み処理 byte型に変換してから書き込む
    s3.upload(
        f"{name}.json",
        json.dumps(_json, ensure_ascii=False).encode('utf-8')
    )



def build_holidays(event):
    """
    googleカレンダーから祝日を取得し、jsonファイルを作成する
    s3に保存されているjsonファイルを取得し、更新する作業まで行う
    :param start_date:
    :param end_date:
    :return:
    """
    now = datetime.now() + relativedelta(years=1) + timedelta(hours=9)
    # 引数なしの場合は、来年の当月を取得するため設定する
    _start_date = _strpdate(event.get('start_date', '')) or now.replace(day=1)
    _end_date = _strpdate(event.get('end_date', '')) or (
        _start_date + relativedelta(months=1) - timedelta(days=1)
    )
    if _start_date > _end_date:
        raise ValueError('start_date must be earlier than end_date')
     = {}
    service = build('calendar', 'v3', developerKey=GOOGLE_API_KEY)
    holiday = request_google_api(
        service,
        CALENDAR_ID,
        _start_date,
        _end_date
    ).get('items', [])
    update_or_create_json("ja", holiday)

request_google_api関数の説明

serviceはbuild_holidays内ですでに呼んでいるので、条件に従ってexcxuteするとdict型でデータを取得できる。
以下にAPIの始め方と取得方法が載っているので、時間あれば見て欲しい。認証方法はOAuthで取得してもいいけど、私はcredentialをコンテナ内にいれたくないなーという観点からいれていない。
補足だが、googleapiclient.discovery内のソースコードをみてもAPIのハシゴ的なモジュールだからあまり得るものはない。書き方とかはdryだからかっこいい書き方をしているのはわかるよ？
参考文献: google calendar api

update_or_create_json関数の説明

S3のバケットにアクセスして、既に保存しているjsonを取り出し、上書きする作業を行なっている。
eventsの内容は、google_apiから取得したデータです。
defaultdictの内容は、個人的に見返してみて違う気がする。

build_holidays関数の説明

mainの関数。
event内には、ポストされたデータが格納されています。
その内容に沿って、googleカレンダーから情報を取得、JSONファイルに保存するようしています。

requestに対してS3のjsonを取得、加工してresponseする

単純にjsonを取得して、欲しいデータを渡す処理を書いています。

import json
import time
from concurrent.futures import ThreadPoolExecutor
from botocore.exceptions import ClientError
from datetime import datetime
from functools import lru_cache

from s3 import S3Client
from utils import dict_flatten, send_notification
import settings


@lru_cache()
def get_s3_holidays(national: str) -> dict:
    """
    s3からjsonファイルを取得する
    キャッシュするが、定期的にs3を更新する際に、キャッシュをクリアする必要がある
    refresh_s3_holidays() でキャッシュをクリアできる
    :param national: 
    :return: 
    """
    print(f'cached: get_holidays({national})')
    try:
        result = S3Client().download(f'{national}.json')
    except ClientError:
        result = b''
    return json.loads(result or '{}')


def validate_params(
    national: str,
    year: int,
    month: int,
    day: int,
) -> None:
    _validate_national(national)
    if year:
        _validate_date(year, month, day)


def _validate_national(national: str) -> None:
    if national not in settings.CALENDAR_IDS.keys():
        raise ValueError(
            f'national(:{national}) must be '
            f'{" or ".join(settings.CALENDAR_IDS.keys())}'
        )


def _validate_date(year: int, month: int, day: int) -> None:
    if year < 2022:
        # 2022年 以上かのチェック（それ以前のデータがないため）
        raise ValueError('year must be greater than 2022')
    # 月と日が正しくないか確認する
    _ = datetime(year, month or 1, day or 1)


def _set_params(event: dict) -> tuple:
    _p = event.get('pathParameters', {})
    return _p.get('national', ''), int(_p.get('year', 0)), int(
        _p.get('month', 0)
    ), int(_p.get('day', 0))


def get_holidays(event: dict) -> list[dict]:
    """
    s3からjsonファイルを取得し、祝日を送信する
    """
    national, year, month, day = _set_params(event)
    validate_params(national, year, month, day)
    s3_holidays = get_s3_holidays(national)
    if year:
        s3_holidays = s3_holidays.get(str(year), {})
    if all([year, month]):
        s3_holidays = s3_holidays.get(str(month), {})
    if all([year, month, day]):
        s3_holidays = s3_holidays.get(str(day), {})
    return dict_flatten(s3_holidays)


def refresh_s3_holidays(event) -> None:
    """
    s3取得関数のキャッシュをクリアする
    クリア後、s3 内の json を再度取得する
    :param event:
    :return:
    """
    start = time.time()
    get_s3_holidays.cache_clear()
    with ThreadPoolExecutor(max_workers=4) as x:
        futures = [
            x.submit(get_s3_holidays, national)
            for national in settings.CALENDAR_IDS.keys()
        ]
    for f, national in zip(futures, settings.CALENDAR_IDS.keys()):
        _ = f.result()
    send_notification(f'refresh_s3_holidays time: {time.time() - start}秒')

get_s3_holidays関数の説明

S3からJSONを取得する。ない場合は空辞書を渡す

validate_params関数の説明

パラメーターの情報のバリデート、国、年、月、日まで指定するできるので、正しい値が入っているか確認する

get_holidaysの説明

国、年、月、日まで指定されている場合、jsonから目的のデータを取得する
国が最低限になっているので、国が指定してない時は落ちる

s3接続用のクライアントクラス

s3.py

import boto3
import io
import settings


class S3Client:
    client = boto3.client('s3')

    def __init__(self, bucket_name: str | None = None):
        self.bucket = bucket_name or settings.DEFAULT_BUCKET

    def download(self, key: str) -> bytes:
        with io.BytesIO() as buffer:
            self.client.download_fileobj(self.bucket, key, buffer)
            buffer.seek(0)
            return buffer.read()

    def upload(self, key: str, data: bytes):
        with io.BytesIO(data) as buffer:
            self.client.upload_fileobj(buffer, self.bucket, key)

AWSの構成

以下のサービスを有効にしていきます。
画面は割愛します（どうでもいいけど割愛の言葉の由来は蚕からきているようですよ）

• route53サブドメインの登録
• API Gatewayの設定
• AWS Lambdaの設定（コンテナイメージを使ってね）

AWSの構成で苦戦したところ

AWS Lambdaの設定当初、s3への接続ができなかった。
「コンテナイメージだし、環境変数を設定しないとだめだよねー」と環境変数使うためにdockerfile内環境変数の設定とiamでユーザーを作ろうとしたが、それはbad choiceだと後々気づく。

ロールを作ってLambdaに割り当てるのがコンテナイメージでも正解だったようです。
ロールを割り当てることで自動的に、AWS Lambda内の環境変数にaws_access_key_idとaws_secret_access_keyを登録してくれる。
その環境変数はコンテナイメージ内でも特に宣言せず自動的に宣言されるようになっている。（その恩恵はLambda用のイメージを継承しているからだが）
そのおかげで、特にアクセスキーなどコード上に記載する必要なく構築できているわけです。（AWSありがとう）

実際に動かす

route53やAPIGatewayで登録した後、実際にAPIを取得してみます。
ある程度JSONファイル内に書き込まれているので以下で取得できるようになっています。
（3年3ヶ月分のデータだが、60件しか登録されてない）

以上

サーバーレスで祝日サーバー（レス）を作ってみました。
比較的簡単に、かつ低コストに作れたのではないかなと思っています。
ただ現状欠点として、ある一定時間呼ばれなかったとき、Lambdaの再構築する時間発生するのでロード時間が長いかも（1〜2秒ぐらいかかる）。
ただ、頻度よく呼ばれ続けるとLambda内でキャッシュされるので処理は格段に早くはなります。
こういう遅延も今後改善していければなと思っています。（Lambdaじゃ無理かも？）