Withings APIの測定値をBigQueryへ取り込む｜Cloud Runで取得する

はじめに

Withings の体重計で測った値を、スプレッドシートの自動化で集めるのをやめ、BigQuery でほかのデータと一緒に見られるようにしたいと考えました。

本記事では、Withings API（開発者向けサイト）で何が取れるのか、連携のやり方にどんな選択肢があるのか、そのうち「自分で取りに行く」を選んだときに何を書くことになるのかを順に説明します。取得は Cloud Run Job で動かし、Cloud Storage を経由して BigQuery まで入れています。

認証とトークンの扱いは、体重計に限らず同じ作りの API で共通して使えます。

本記事のまとめ

• 連携のやり方は 3 つあり、保存先と形を自分で決めたいなら「自分で取りに行く」を選ぶことになる。その代わり、認証とトークンの面倒は全部自分で引き受ける。
• アクセストークンの更新は、client_secret をそのまま送らず、それを鍵にした署名を送る形でも呼べる。レスポンスの status=503 はサーバーの不調ではなく、送ったパラメータが不正という意味である。
• 更新のたびに入れ替わるトークンは、新しい値を保存してからアクセストークンを使う。使った時点で古い値が失効するので、先に使って保存に失敗すると認可のやり直しになる。
• 測定値はそのままの数字では返らず、整数と 10 の指数の組で届く。BigQuery へは文字列のまま運び、型を決めるのは最後にした。

Withings APIで取れるものと、取る前に要るもの

取れるのは 1 回の測定ごとにまとまった値で、体重だけでなく体脂肪率や筋肉量も同じかたまりに入っています。測定のかたまりには ID が振られていて、あとで重複を取り除くときの手がかりになります。

値の種類は名前ではなく番号で区別します。体重、除脂肪量、体脂肪率、体脂肪量、筋肉量、水分量、推定骨量といった具合で、測りたい種類を番号で指定して取りに行きます。測られなかった種類は、空の値ではなく最初から配列に入ってきません。これが後で効いてくるので、覚えておいてください。

取りに行く前に要るものは 3 つです。

1. 開発者向けの登録をして、クライアント ID とクライアントシークレットを受け取る
2. ブラウザで 1 回だけ利用者の認可を通し、refresh token を受け取る
3. 以降は、その refresh token からアクセストークンを作って API を呼ぶ

1 回目の認可だけは手で通す必要があり、ここは自動化できません。2 回目以降は 3 の繰り返しになります。

呼び出しの間隔について、公式の資料は 1 ユーザーあたり 10 分に 1 回より多くポーリングしないよう求めています。上限を超えると、本文の status が 601 になります。私の使い方は 1 日 2 回なので、問題になっていません。

連携のやり方は3つある

測定値を自分のところへ持ってくる方法は 3 つあり、どこまで自分で作って管理するかが違います。

私は 3 つ目にしました。それまではスプレッドシートの自動化で集めていたのですが、トークンと API キーの置き場所、テストのしやすさ、構成をコードで管理できることの 3 点で切り替えています。

通知は、公開する受け口と購読の管理、再送への対応が要るので見送りました。以降は 3 つ目を選んだ場合の話になります。自分で書くかツールに任せるかの見分け方は、dltHubが引き受けるETL処理の5つに書きました。

トークンの更新は署名の形でも呼べる

アクセストークンの更新には、client_secret をそのまま送る形と、client_secret を鍵にした HMAC-SHA256 の署名を送る形があります。公式のリファレンスにどちらの形も載っています。私の実装は署名の形で、client_secret をそのまま送る形は試していません。

HMAC は、あらかじめ共有した秘密の値を鍵にして、文字列から固定長の値を計算する仕組みです。受け取った側が同じ計算をして一致を見るので、送信元を確かめられます。

署名を付けてトークンを更新するには、API を 2 回呼びます。1 回目で受け取る nonce は、1 回だけ使える使い捨ての値です。

1. nonce を受け取る呼び出しを 1 回する。この呼び出し自体も、処理名・クライアント ID・UNIX 時刻の 3 つを署名して投げる
2. 返ってきた nonce を使って、トークン更新の呼び出しを署名する

nonce を使うのは、盗み見られた同じリクエストを再送されても通らないようにするためです。公式の 署名の手引きも「The nonce is a transient token, and every request requires a fresh one」と書いているので、使い回さず毎回取り直します。

署名する文字列は、パラメータ名をアルファベット順に並べたときの値だけをカンマでつないだものです。パラメータ名は入りません。実装は次のようになりました。

def sign_request(params: dict[str, Any], client_secret: str) -> str:
    """Withings Signature v2 用の HMAC SHA-256 signature を作成する。"""
    # Withings の署名対象は action, client_id と nonce または timestamp の値をカンマ結合した文字列。
    values = [str(params["action"]), str(params["client_id"])]
    if "nonce" in params:
        values.append(str(params["nonce"]))
    elif "timestamp" in params:
        values.append(str(params["timestamp"]))
    else:
        raise WithingsApiError("Withings signature params require nonce or timestamp.")

    message = ",".join(values).encode()
    return hmac.new(client_secret.encode(), message, hashlib.sha256).hexdigest()

トークン更新の本体には refresh_token と署名を入れますが、client_secret は入れません。うっかり本体に混ぜるとログやプロキシに残るので、テストでも入らないことを確かめています。測定値を取る呼び出しでは、アクセストークンを Authorization ヘッダーにだけ載せています。

レスポンスのstatus=503はHTTPの503ではない

HTTP は 200 で返るのに、本文の status が 0 以外になることで失敗を伝える API でした。しかも 503 はサーバー側の不調ではなく Invalid params を意味します。HTTP のステータスだけを見ていると、失敗したことに気づけません。

そこで、すべてのレスポンスで status を検査して 0 以外なら例外にする処理を 1 箇所にまとめました。503 が返ったときに疑うのは、サーバーではなく自分が送ったパラメータです。認可コードの期限切れ、リダイレクト先の不一致、nonce と署名の不備が候補になります。

初回の認可で引っかかりやすい点は 3 つです。

• 認可 URL に state を付けないと先に進めない。state は認可のリクエストと戻りを突き合わせるための値である
• 認可コードの有効期間は 30 秒しかない。ブラウザから値をコピーして手元のコマンドに貼る間に切れる
• 戻り先を受けるサーバーを立てていなくてもよい。ブラウザが接続エラーの画面になっても、アドレスバーに code が入っていればそれを使える

state を忘れたときに返ってきたのは次の内容です。

error=invalid_request
The state parameter is required

Cloud Run Jobで呼ぶときの中身

この Job がするのは、API を呼んで、返ってきた測定値を 1 行 1 JSON のテキストとして Cloud Storage に置くことだけです。BigQuery に入れるのも、整形して集計するのも、別の Job に分けました。

Cloud Run には Service と Job があります。Service はリクエストやイベントを受けて動き続ける形、Job は呼び出されたときに始まって終わる形です。公式ドキュメントの resource model では、Job の実行は「job execution」として作られ、その中で task が走ると整理されています。決めた間隔で API を取りに行って終わる処理は Job 側です。

1 行 1 JSON のテキストは NDJSON と呼ばれる形式で、改行で区切られた JSON が並びます。BigQuery はこの形式をそのまま読めるので、取得側は整形を考えずに書き出せます。

責務をここで切った効果は、依存しているライブラリに出ました。この Job に入っているのは requests、Secret Manager のクライアント、Cloud Storage のクライアントの 3 つだけで、BigQuery のクライアントは入っていません。権限も同じで、この Job が使うサービスアカウントは BigQuery に 1 行も書けません。取得が失敗してもテーブルは壊れず、集計を直したいときに取得のコードを読まなくて済みます。

過去にさかのぼって取り直すときは、Job の引数ではなく環境変数の上書きで日数を渡しています。引数にすると、取り直し用の設定が毎回必要なものに見えてしまうためです。

次の画面は、同じ作りで動かしている変換側の Job の実行履歴です。取得側ではありませんが、1 回の実行がどれくらいで終わるのかはここで見ます。

入れ替わるトークンは保存してから掃除する

この API は更新のたびに refresh token が入れ替わるので、新しい値を保存してから、新しいアクセストークンを使います。古い値は、発行から 8 時間後か、新しいアクセストークンを使った時点で失効します（公式の説明）。

トークンの保存先は Secret Manager です。1 つのシークレットに対して値を追加していく作りで、追加するたびに version という単位で積まれていきます。更新するのではなく追加するので、古い値も残ります。

# Withings の refresh token は rotate されるため、新しい access token を使う前に必ず保存する。
if token.refresh_token != refresh_token:
    secret_store.add_secret_version(config.withings_refresh_token_secret, token.refresh_token)

    # 保存できた後に古い version を掃除する。
    # 掃除の失敗はデータ取得の失敗ではないため、パイプライン全体は止めない。
    try:
        pruned = secret_store.prune_secret_versions(config.withings_refresh_token_secret)
    except Exception as error:  # noqa: BLE001 - 掃除の失敗で本処理を落とさない
        ...

保存に失敗したら、そこで Job を止めます。新しいアクセストークンをまだ使っていなければ、古い値は最大 8 時間使えるので、ブラウザでの認可をやり直さずに済む可能性があります。先に測定値を取りに行って保存に失敗すると、古い値が失効して新しい値も手元に残らず、認可のやり直しになります。

この順序はテストで固定しました。呼び出された順を記録して ["refresh", "add_secret_version", "prune_secret_versions", "get_measure_groups", ...] の並びと比較しているので、実装で入れ替えるとテストが落ちます。コメントで注意を書くだけだと、あとから触る人（半年後の自分を含みます）が気づかずに並べ替えます。

破棄が取り消せないことも効いています。公式ドキュメントは、破棄された version の状態は「cannot be changed to another state」と書いています。破棄しても困らないのは、すでに失効した古い version です。

破棄する処理を入れる前は、実行した回数だけ version が積み上がっていました。公式ドキュメントによると無効にした version にも課金されるので、無効にするだけでは足りません。今は最新の 3 版だけを残し、それより古い version を破棄しています。消して困るのは最新の 1 つだけですが、誤って消したときの猶予として 3 版にしました。

測定値の運び方と、BigQueryに入るまで

取った値をそのまま BigQuery に書き込まず、Cloud Storage に NDJSON で置いてから読ませています。取得と取り込みを分けておくと、取り込みをやり直したいときに API を呼ばずに済みます。

置き場所のパスは次の形です。

def build_body_measurements_object_name(*, fetched_date: str, prefix: str = "body_measurements") -> str:
    """Withings 測定値 JSONL の GCS object path を組み立てる。"""
    # fetched_dt は測定日ではなく Job 実行日。同日再実行では latest を上書きする。
    normalized_prefix = prefix.strip("/")
    return f"{normalized_prefix}/fetched_dt={fetched_date}/body_measurements_latest.jsonl"

fetched_dt は測定日ではなく Job を動かした日です。同じ日に 2 回動けば、2 回目が 1 回目のファイルを上書きします。取り直しても増えないので、失敗したら何も考えずにもう一度動かせます。0 件のときは書きません。空のファイルで上書きすると、後ろの取り込み側がその日のデータを失うからです。

どこまで取ったかを記録するファイルは持たず、毎回「今から 30 日前まで」を取り直しています。測定値は後から変わるからです。アプリ側の同期が遅れて届くことがあり、過去の測定を手で直すこともあります。前日分だけを取る作りにすると、それらが永久に反映されません。

BigQuery からは external table として読みます。external table は、ファイルをコピーせずテーブルとして読む仕組みです。パスの fetched_dt= という書き方は Hive パーティションと呼ばれ、ディレクトリ名がそのまま列になります。ここで階層を 1 つだけにしたのには、3 つ理由があります。

• BigQuery に渡す URI には、ワイルドカードを 1 個しか書けない。しかも * は / もまたぐので、階層が増えると 1 個では言い表せなくなる
• パーティションの前置きは、パーティションのキーの直前で終わらせる必要がある。途中に種別のディレクトリを挟めない
• パーティションのキーと同じ名前の列をファイルの中にも持つと、external table の作成が失敗する。fetched_dt はパスにだけ出し、NDJSON の中には書いていない

external table は測定値の列を文字列で受け、数値への変換はその先のテーブルで行っています。取り込みの入口で型を決めてしまうと、1 件だけ想定外の値が来たときに全体が止まるためです。重複は測定のかたまりの ID を手がかりに取り除き、同じ ID が複数来たときは更新時刻が新しいほうを採ります。

値はvalueとunitの組から組み立てる

この API は測定値をそのまま返さず、整数と 10 の指数の組で返します。実際の値は value に 10 の unit 乗を掛けたものです。value が 71500 で unit が -3 なら 71.5 になります。

def decode_measure_value(measure: dict[str, Any]) -> str:
    """Withings の value / unit 表現を実測値の文字列へ変換する。"""
    # Withings は value と 10^unit の組み合わせで実値を表す。
    value = Decimal(str(measure["value"]))
    unit = int(measure["unit"])
    decoded = value * (Decimal(10) ** unit)
    return _decimal_to_string(decoded)

float を通さず Decimal で計算し、末尾のゼロを落とした文字列で書き出しています。体組成の値は小数第 1 位まで意味があるので、途中で丸めを挟みたくありませんでした。

もう 1 つ、配列の順番で値を取らないようにしました。前のほうで書いたとおり、測られなかった種類は配列に入ってきません。1 番目を体重、2 番目を体脂肪率と決め打ちすると、体脂肪率が測れなかった日に全部ずれます。種類を表す番号で判定して列に振り分けてください。

まとめ

Withings API を自分で呼ぶと決めた時点で、面倒の大半は認証とトークンに集まります。測定値を取って保存する部分そのものは、拍子抜けするほど短く書けました。

• アクセストークンの更新を署名の形で呼ぶなら、nonce は毎回取り直す。レスポンスは HTTP ではなく本文の status を見る
• refresh token は更新のたびに入れ替わる。新しい値を保存してからアクセストークンを使い、順序はテストで固定する
• 古い version を破棄する処理を入れないと、実行した回数だけ積み上がる。無効にしただけでは課金は止まらない
• 測定値は整数と指数の組で届き、測られなかった種類は配列に入ってこない。位置ではなく種類の番号で振り分ける

これから同じことをするなら、連携のやり方 3 つのどれを採るかを先に決めてください。つなぎ先がすでにあるなら 1 つ目で終わりますし、保存先と形を自分で決めたいなら、この記事で書いた面倒を引き受けることになります。

なお、Cloud Run の既定値や上限、Secret Manager の課金対象、Withings の開発者向けページの構成は変わります。実際に設定するときは、最新を公式ドキュメントで確認してください。