BigQuery Remote Functionsを使ってmicroCMSのデータをBigQueryに転送する

microCMSで管理されているデータをBigQuery Remote Functionsを使ってBigQueryにデータ連携してみます。
基本的な考え方は以下のブログで紹介しているので、まずはこちらをごらんください。

読み出し用のPython関数

今回はPythonを使ってCloud Functionsで動く関数を作ってみます。
microCMSには様々な言語・フレームワーク用のSDKがあるようですが、Python用のものはないので、requestsライブラリでREST APIを叩いてみます。

以下の関数をCloud Functionsにデプロイし、BigQueryから読み出すためのconnectorなどもデプロイします。
詳細な方法は冒頭で紹介したブログ記事を参照ください。

import os
import logging

import functions_framework
import requests
import flask

def read_microcms_records(api_endpoint, api_key, batch_size=100):
    microcms_subdomain = 'microCMSのサブドメインを入れる'
    microcms_domain = f'{microcms_subdomain}.microcms.io'
    api_endpoint = f"https://{microcms_domain}/api/v1/{api_endpoint}"
    headers = {
        "Content-Type": "application/json",
        "X-MICROCMS-API-KEY": api_key,
    }

    raw_records = []

    offset = 0
    while True:
        params = {"limit": batch_size, "offset": offset}
        logging.info(f"GET: {api_endpoint} {params}")
        result = requests.get(api_endpoint, headers=headers, params=params)

        if result.status_code == 200:
            response = result.json()
            raw_records.extend(response['contents'])
            total_count = response['totalCount']

            logging.info(f"total count is {total_count}")
            logging.info(f"{len(raw_records)} rows fetched")

            offset += batch_size
            if total_count == len(raw_records):
                break
        else:
            logging.error(result.status_code)
            raise RuntimeError(f"Error while reading microCMS records: {result.status_code}")
            break

    return raw_records

@functions_framework.http
def read_microcms(request):
    api_key = os.environ['MICROCMS_API_KEY']
    try:
        request_json = request.get_json()
        calls = request_json['calls']
        if len(calls) != 1:
            raise RuntimeError("this function must be call in scalar subquery!")

        api_endpoint = calls[0][0]
        records = read_microcms_records(api_endpoint, api_key)
        return_value = [records]

        return flask.make_response(flask.jsonify({"replies": return_value}))
    except Exception as e:
        return flask.make_response(flask.jsonify( {"errorMessage": str(e)}), 400)

あとは、以下のようなSELECT文を実行すればmicroCMSのデータをBigQueryで読み取ることができます。

SELECT `プロジェクトID.データセット名.read_microcms(<APIのエンドポイント>)`