💡
Looker API を使用してダッシュボード配信

2025/04/30に公開
 背景私たちのチームでは、Lookerで作成したダッシュボードを、Lookerアカウントを持たない多くのユーザーにも定期的に共有する必要がありました。これらのダッシュボードには、各ユーザーにとって重要な特定のフィルタリングが適用されている必要がありました。

 問題点対象ユーザーが非常に多かったため、Lookerの標準機能であるスケジュール設定を手動で一人ひとりに行うのは、現実的ではありませんでした。設定ミスや漏れの可能性も高く、運用負荷の増大が懸念されました。

 実施概要この問題を解決するために、以下の手順でLookerダッシュボードのメール自動配信システムを構築しました。
ユーザーごとの配信先メールアドレスとダッシュボードのフィルタ条件をまとめたcsvファイルの作成
Looker API を使用して、フィルタをかけた状態でダッシュボードをPDF化する
取得したPDFをユーザーにメールで配信する
今回は上記の2のLooker API について詳しく記載をしていきます。

 前提知識
 Looker APIの概要Looker APIは、Lookerの様々な機能にプログラムからアクセスするためのインターフェースです。RESTfulなAPIとして提供されており、HTTPリクエストを送信することで、Lookerのデータやオブジェクト（ダッシュボード、Look、ユーザーなど）の操作、レポートのエクスポートなどを自動化できます。
Looker APIを利用することで、以下のようなことが可能になります。

データの抽出と加工: LookerのExploreで定義されたデータをプログラムから取得し、必要な形式に加工して外部システムで利用する。

レポートのエクスポート: ダッシュボードやLookをPDF、CSV、JSONなどの形式でエクスポートする。

Lookerオブジェクトの管理: ダッシュボードやLookの作成、更新、削除、ユーザーや権限の管理などを自動化する。

外部システムとの連携: 他のBIツールやデータ分析基盤、業務システムなどとLookerを連携させる。
注釈: Looker APIを利用するためには、事前にLookerの管理者からAPIキー（クライアントIDとクライアントシークレット）を発行してもらう必要があります。APIキーの発行手順については、Lookerの管理ドキュメントをご確認ください。

 Looker API SDKの活用Looker APIを直接利用することも可能ですが、より効率的に開発を進めるために、Lookerが提供しているLooker API SDKを利用することが推奨されます。
Looker API SDKは、様々なプログラミング言語（Python、Ruby、TypeScript、Javaなど）に対応したライブラリの集合であり、Looker APIの各エンドポイントへのリクエスト送信やレスポンスの処理を抽象化し、開発者がよりビジネスロジックに集中できるよう支援します。
今回のダッシュボードPDF化処理では、Python版のLooker API SDKを活用しました。Python SDKを利用することで、Looker APIへの認証、ダッシュボードの指定、フィルタ条件の設定、PDF形式でのレンダリング指示といった一連の処理を、簡潔かつ可読性の高いコードで実装することができました。

 具体的な実施内容基本的にはLookerのSDKサンプルGitHubリポジトリに公開されているダッシュボードPDFダウンロードのコードを流用しています。
pypiから最新のlooker_sdkをインストールします。

※ローカルマシンで行う場合、pyenvを使って環境を管理することをお勧めします。
pip install looker_sdk
ダッシュボードからPDFを取得する
download_dashboard_pdf.py
import os
import json
import urllib
import sys
import time
from typing import cast, Dict, Optional
import looker_sdk
from looker_sdk import models40 as models
# 便宜上ベタ打ちしていますが、セキュリティのため、これらの情報は環境変数として設定することを強く推奨します
url = "https://your_looker_instance.com:19999"
client_id = "your_client_id"
client_secret = "your_client_secret"

dashboard_title = "dashboard_title" #PDFファイルの名前に使用されます
dashboard_id = "1" #LookrtダッシュボードのIDを指定します sample:https://your_looker_instance/dashboards/1
filters = json.loads('{"filter1": "value1, value2", "filter2": "value3"}') #ダッシュボードで使用したいフィルタ条件を指定します
pdf_width = "1500"
pdf_height = "4000"
pdf_style =  "tiled"

def init_looker_sdk(url: str,client_id: str,client_secret: str) -> looker_sdk.methods40.Looker40SDK:
   os.environ["LOOKERSDK_BASE_URL"] = url
   os.environ["LOOKERSDK_CLIENT_ID"] = client_id
   os.environ["LOOKERSDK_CLIENT_SECRET"] = client_secret
   os.environ["LOOKERSDK_VERIFY_SSL"] = "true" 
   os.environ["LOOKERSDK_TIMEOUT"] = "120"
   os.environ["LOOKERSDK_API_VERSION"] = "4.0"
   return looker_sdk.init40()

#looker_sdkを初期化
sdk = init_looker_sdk(url, client_id, client_secret)

def main():
    dashboard = cast(models.Dashboard, get_dashboard(dashboard_id))
    download_dashboard(dashboard, pdf_style, pdf_width, pdf_height, filters)

def get_dashboard(dashboard_id: str) -> Optional[models.Dashboard]:
    """Get a dashboard by id."""
    dashboard = next(iter(sdk.search_dashboards(id=dashboard_id)), None)
    if not dashboard:
        raise Exception(f'dashboard "{dashboard_id}" not found')
    return dashboard

def download_dashboard(
    dashboard: models.Dashboard,
    style: str = "tiled",
    width: int = 545,
    height: int = 842,
    filters: Optional[Dict[str, str]] = None,
    ):

    """Download specified dashboard as PDF"""
    id = dashboard.id
    task = sdk.create_dashboard_render_task(
        id,
        "pdf",
        models.CreateDashboardRenderTask(
            dashboard_style=style,
            dashboard_filters=urllib.parse.urlencode(filters) if filters else None,
        ),
        width,
        height,
    )

    if not (task and task.id):
        raise Exception(
            f'Could not create a render task for "{dashboard.title}"'
        )

    # poll the render task until it completes
    elapsed = 0.0
    delay = 0.5  # wait .5 seconds
    while True:
        poll = sdk.render_task(task.id)
        if poll.status == "failure":
            print(poll)
            raise Exception(
                f'Render failed for "{dashboard.title}"'
            )
        elif poll.status == "success":
            break

        time.sleep(delay)
        elapsed += delay
    print(f"Render task completed in {elapsed} seconds")

    result = sdk.render_task_results(task.id)
    filename = dashboard_title + ".pdf"
    with open(filename, "wb") as f:
        f.write(result)
    print(f'Dashboard pdf saved to "{filename}"')

main()
このコードを実行することで、指定したLookerダッシュボードを指定したフィルタ条件でPDFファイルとしてローカルに保存できます。
ダッシュボードのPDFを取得できるようになったので、あとは以下の処理を実装することで、メール自動配信システムが完成します。
フィルタ条件の動的変更処理: CSVファイルから読み込んだユーザーごとのフィルタ条件を、PDF生成処理に反映させる。

メール送信処理: 生成されたPDFファイルを、CSVファイルに記載されたメールアドレス宛に送信する。

これらの処理を実装することで、大量のユーザーに対するダッシュボードのメール配信を自動化し、効率的な情報共有を実現が可能です。

 最後に今回の開発では、Looker API SDKを活用することで、Lookerのダッシュボードをプログラムから操作し、PDFとして取得できることを確認しました。
課題として、このPDF作成処理はユーザー数に応じて処理時間が増大するため、実運用においては並列処理などの最適化が不可欠となります。

並列処理について学習を進めており、今後の実装に取り入れていく予定です。プログラム上で並列処理機能を活用することで、効率的なメール配信システムの構築を目指します。
背景

問題点

実施概要

前提知識

Looker APIの概要

Looker API SDKの活用

具体的な実施内容

最後に

Discussion
