定期的にAPIから取得したデータをCloud Storageへ格納にしてみた

お久しぶりです。株式会社KIYONOでエンジニアをやっております西園と申します。

今回はAPIを自動取得し、取得したデータをCloud Storageへ保存する仕組みを解説したいと思います。

構成図
Cloud Storageの設定
Cloud Functionsの設定
Cloud Schedulerの設定
起動
まとめ

構成図

上図のように今回は無料で使用できる郵便番号検索APIを使用します。このAPIは郵便番号をパラメータで送るとその住所が返ってくるAPIです。

Cloud Storageに保存する流れとしては、下記になります。

Cloud Schedulerで設定したタイミングでCloud FunctionsへHTTP通信を行い関数を実行させる。
郵便番号検索APIにGETリクエストを送信し、データを取得する。
APIから取得したデータをCloud Storageへ保存する。

Cloud Storageの設定

まずは保存先に使用するバケットを作成します。

上記のように作成していますが、設定は各要件に合う形で設定すれば良いです。

※Cloud FunctionsでCloud Storageのバケットを作成するコードを記載しても実装できますが、下記懸念点があり、あらかじめ作成しておくことをおすすめします。

バケット名は世界中で一意である必要があるため、作成する際にバケット名が被ってしまうと作成されない。
バケットが被った際のハンドリングを実装しないといけないので工数が増える。

Cloud Functionsの設定

次にCloud Functionsの設定を行います。

基本設定

まずは、初期の設定は下記のようにします。

関数名：任意の名前を入力
リージョン：任意のリージョン
トリガー：今回はHTTPリクエストに対して関数を発火させるのでHTTPに設定します。
認証：本来はIAMを設定してロールを持ったリソースからのみ呼び出しを許可する必要あり、今回は簡素化のために未承認でも許可しています。
HTTPS：必須にする場合はチェックを入れる。（今回は必要ないので外しています。）

コードの実装

今回は、python3.10を使用して記載していきます。

まず、エントリポイントを設定します。

ここには、Cloud Functionsで起動する関数名を記載してください。

今回はmain()を作成していきますのでmainで設定しています。

main.py

from google.cloud import storage
import requests

from datetime import datetime


BASE_URL = 'https://zipcloud.ibsnet.co.jp/api/search'
KIYONO_ZIPCODE = '1510051'
BUCKET_NAME = 'nishizono-tech-blog-test'


def main(request):
    params = {
        'zipcode': KIYONO_ZIPCODE,
    }
    api_contents = get_api_contents(params)

    upload_to_storage(
        KIYONO_ZIPCODE,
        BUCKET_NAME,
        api_contents
    )

    print('Finish upload to storage')

    return 'OK'


def get_api_contents(params):
    response = requests.get(
        url=BASE_URL,
        params=params
    )

    return response.text


def upload_to_storage(zip_code, bucket_name, contents):
    client = storage.Client()
    now = datetime.now()
    file_name = f'{zip_code}_{now}.json'
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    blob.upload_from_string(
        date=contents,
        content_type='application/json'
    )

requirements.txt

google-cloud-storage==2.4.0
requests==2.28.1

ちなみにAPIでトークンなどの機密情報を扱う場合は、平文記載することはセキュリティの面で危険なので下記のようにランタイム環境変数を設定し、osモジュールのgetenv()メソッド等で呼び出すことをおすすめします。

動作確認

Cloud Functionsの設定が終わりましたら、実装した内容がしっかり動作するかを確認しましょう。

作成したCloud Functions詳細画面へ行き、テスト中のタブをクリックすると以下の画面に行くと思います。

関数をテストするボタンを押すと関数が実行されてページ下部にログが表示されます。

ここでしっかり動作していたらチェック完了です。

Cloud Schedulerの設定

最後にCloud Schedulerの設定を行います。

まずはいつ起動するかを設定します。

頻度の部分で起動タイミングを指定できます。今回は毎時5分（12:05, 13:05…）に取得するように指定します。

記載の仕方は下記のようになります。

基本型：* * * * *

それぞれのアスタリスクの意味は下記のように考えてください。

分時日月曜日

例えば、

毎日11時30に起動したい場合

30 11 * * *

毎月1日の12時15分に起動したい場合

15 12 1 * *

毎週月曜日の15時45分に起動したい場合

45 15 * * 1

※日曜日0~土曜日6

次にHTTP通信を送るURLを設定します。

URLについてはCloud Functionsのトリガーで確認することができます。

以上で設定は終わりになります。

起動

では実際に起動しているか確認してみましょう。

まずCloud Schedulerが起動しました。

次にCloud StorageにAPIの情報が入っているかを確認するために作成したバケットを見に行きます。

しっかりJSONファイルが作成されていますね。

最後に取得できた中身を見てみましょう。

しっかり住所も取得できています。

まとめ

いかがだったでしょうか。今回は指定した郵便番号の住所を定期的に取得し、ストレージに格納するという簡単な構成の紹介をしました。

実務でこの構成を使うとしたら、ECサイトなどの日々更新されるようなデータを持っているAPIに対して日時でバッチ処理をしてデータを取得するといったことに使用できると思います。

また今回の構成を拡張してBigQueryまでデータ連携すれば、分析までできる基盤を構築できそうですね。

ぜひみなさんも試してみてください。