estat-api-dlt-helper 0.2.0

e-Stat APIを使ってデータを取得し、dltを使ってデータをロードするためのヘルパーライブラリ

estat_api_dlt_helper

e-Stat APIからデータを取得しロードするhelper

ドキュメントはこちら

概要

e-Stat APIを利用してデータを取得し、DWHなどのデータ基盤にロードするシーンでの活用を想定しています。

Pythonのライブラリとして動作し、以下の機能を提供します。

• parse_response
  • APIのレスポンスをパースし、データとメタデータを結合させたArrow Tableを作成します。
• load_estat_data
  • dlt(data load tool)のラッパーとして動作し、 統計表IDとテーブル名などを設定するだけで、簡単にDWHなどにロード可能です。
  • paginationや複数の統計表IDを同じテーブルにロードしたいケースなどを内部でいい感じに処理します。
• create_estat_resource / create_estat_pipeline / create_estat_source
  • dltのresource・pipeline・sourceをそれぞれ個別に生成し、dltの細かい設定や機能を使いこなしたい場合に使用します。
  • create_estat_sourceは複数のEstatDltConfigをまとめてdlt sourceを生成できます。

モチベーションとコンセプト

それなりの数の政府統計の統計表を効率よくデータ基盤に抽出・ロードしたいというニーズをもとに生まれました。 e-Stat APIのレスポンスはある程度抽象化されているため、メタデータを本体データに結合するパーサーと、 データロードスクリプトを非常に抽象化・量産化できるdlt(data load tool)を組み合わせることで、上記を達成できると感じて実装を始めました。

コンセプト:

• なるべく統計表IDとテーブル名を記述するだけで動くものがいい
• 複数の統計表IDのロードや、マージ戦略などの設定にも対応したい
• 何のデータソース(統計表ID)を、どこに(DWH|データセット|テーブル)、ロードするか、という設定をなるべく同じところで記述したい
• どの統計表のレスポンスにも対応できるパーサーが欲しい

インストール

pip install estat_api_dlt_helper

# BigQuery
pip install "estat_api_dlt_helper[bigquery]"

# Snowflake
pip install "estat_api_dlt_helper[snowflake]"

# duckdb
pip install "estat_api_dlt_helper[duckdb]"

使用方法

e-Stat APIに関して、ユーザー登録やアプリケーションIDの取得が完了している前提です。 取得したアプリケーションIDは環境変数に入れておいてください。

export ESTAT_API_KEY=YOUR_APP_ID

Win:

$env:ESTAT_API_KEY = "YOUR_APP_ID"

parse_responseの使い方

e-Stat APIの/rest/3.0/app/json/getStatsDataのレスポンスをparse_response()に渡すことで、 responseのTABLE_INF.VALUEの中身をテーブルとして、CLASS_INF.CLASS_OBJの中身をメタデータとして名寄せさせたArrow Tableを生成することができます。

処理イメージ:

response	加工後

see: examples

import os
import pandas as pd
import requests

from estat_api_dlt_helper import parse_response

# API endpoint
url = "https://api.e-stat.go.jp/rest/3.0/app/json/getStatsData"

# Parameters for the API request
params = {
    "appId": os.getenv("ESTAT_API_KEY"),
    "statsDataId": "0000020201",  # 社会人口統計 市区町村データ 基礎データ
    "cdCat01": "A2101",           # 住民基本台帳人口（日本人）
    "cdArea": "01100,01101",      # 札幌市, 札幌市中央区
    "limit": 10
}
try:
    # Make API request
    response = requests.get(url, params=params)
    response.raise_for_status()
    data = response.json()
    # Parse the response into Arrow table
    table = parse_response(data)
    # Print data
    print(table.to_pandas())

except requests.RequestException as e:
    print(f"Error fetching data from API: {e}")
except Exception as e:
    print(f"Error processing data: {e}")

load_estat_dataの使い方

dlt(data load tool)のwrapperとして簡便なconfigで取得データを DWH等にロードできます。

ロード可能なDWHについてはdltのドキュメントを参考にしてください。

see: examples

# duckdbの場合
import os
import dlt
import duckdb
from estat_api_dlt_helper import EstatDltConfig, load_estat_data

db = duckdb.connect("estat_demo.duckdb")

# Simple configuration
config = {
    "source": {
        "app_id": os.getenv("ESTAT_API_KEY"), #(必須項目)
        "statsDataId": "0000020201",  # (必須項目) 人口推計
        "limit": 100,  # (Optional) 1 requestで取得する行数 | デフォルト:10万
        "maximum_offset": 200,  # (Optional) 最大取得行数
    },
    "destination": {
        "pipeline_name": "estat_demo",
        "destination": dlt.destinations.duckdb(db),
        "dataset_name": "estat_api_data",
        "table_name": "population_estimates",
        "write_disposition": "replace",  # Replace existing data
    },
}
estat_config = EstatDltConfig(**config)

# Load data with one line
info = load_estat_data(estat_config)
print(info)

load_estat_dataの使い方 (Advanced)

load_estat_data()は簡単な設定でロードを可能にしますが、dltの細かい設定や機能を使いこなしたい場合(dlt.transformやbigquery_adapterなど)は、 dltのresourceとpipelineをそれぞれ単体で生成し、既存のdltのコードと同じように扱うこともできます。

see: examples (resource)

see: examples (pipeline)

create_estat_sourceの使い方

複数の統計表を一括でロードしたい場合は、create_estat_source()を使用できます。 EstatDltConfigのリストを渡すことで、複数のリソースをまとめたdlt sourceを生成します。

see: examples

import os
import dlt
from estat_api_dlt_helper import EstatDltConfig, create_estat_source

app_id = os.getenv("ESTAT_API_KEY")

configs = [
    EstatDltConfig(
        source={"app_id": app_id, "statsDataId": "0000020201"},
        destination={"destination": "duckdb", "dataset_name": "estat", "table_name": "population"},
    ),
    EstatDltConfig(
        source={"app_id": app_id, "statsDataId": "0004028584"},
        destination={"destination": "duckdb", "dataset_name": "estat", "table_name": "household",
                     "write_disposition": "replace", "primary_key": None},
    ),
]

source = create_estat_source(configs)

pipeline = dlt.pipeline(
    pipeline_name="estat_multi",
    destination="duckdb",
    dataset_name="estat",
)
info = pipeline.run(source)
print(info)

Development

# Install development dependencies
uv sync

# Run tests
uv run pytest

# Format code
uv run ruff format src/