日本卸電力取引所のエリアプライスを返すAPIをFastAPIを使用してCloud Functionsにデプロイした話 #Python

はじめに

↓前回の記事

当記事の内容

当記事では、前回の記事でCloud Firestoreにアップロードした日本卸電力取引所（JEPX）のシステムプライス・エリアプライスのデータを取得するためのWeb APIを、PythonのWebフレームワークであるFastAPIを使用して開発した。また、このWeb APIはGoogle Cloud Functionsの第二世代を用いて稼働させている。本記事では、その開発手順を紹介する。

データ

Firestore上の日付毎のシステムプライス・各エリアプライスは以下ファイル構成となっている。

electricity_market_price/    # collection
└── 2023-12-01/    # document
    └── system/    # subcollection
        └── {'00:00':'13.6', '00:30':'13.92', ..., '23:30':'12.39'}    # field
    └── hokkaido/
        └── {'00:00':'13.15', '00:30':'14.0', ..., '23:30':'11.8'}
    └── ...

コンソール画面

システムプライス・各エリアプライスを格納するsubcollection名は以下の通り。

システムプライス : sytem
エリアプライス
- 北海道 : hokkaido
- 東北 : tohoku
- 東京 : tokyo
- 北陸 : hokuriku
- 中部 : chubu
- 関西 : kansai
- 中国 : chugoku
- 四国 : shikoku
- 九州 : kyushu

設計

API名

電源標準単価取得API

概要

指定した日付・エリアにおける電源標準単価を返すAPIとする。

URL

https://＊＊＊＊＊/api/electricity_market_price

メソッド

GET

パラメータ

パラメータ	必須	説明	例
date	Yes	検索日付。フォーマットはyyyy-mm-ddとする。	2023-04-20
area	No	検索エリア。カンマ区切りで複数指定可能。未指定の場合は全エリアを対象とする。	tokyo,kansai

レスポンス

レスポンスはいずれもjsonで返される。

正常時レスポンス

HTTPステータスコード: 200

{
    エリア名: {エリアプライス},
}

例：

{
    "tokyo":{"00:00":"6.5","00:30":"4.92","01:00":"4.88",...,"23:30":"7.8"},
    "kansai":{"00:00":"6.5","00:30":"4.92","01:00":"4.88",...,"23:30":"7.8"},
}

エラーレスポンス

以下形式でレスポンスを返す。内容はエラー一覧を参照のこと。

{
    "detail":{
        "type":エラータイプ,
        "message":エラーメッセージ
    }
}

エラー一覧

No.	ステータスコード	エラータイプ	エラーメッセージ	説明
1	400	MISSING_REQUIRED_PARAMETER_DATE	Date parameter is missing.	リクエストのパラメータにdateが含まれていない場合
2	400	INVALID_DATE_FORMAT	Invalid date format. Please specify the date in yyyy--mm--dd format.	日付のフォーマットが不正な場合
3	400	INVALID_AREA_PARAMETER	Invalid area parameter. Allowed values are "system", "hokkaido", "tohoku", "tokyo", "hokuriku", "chubu", "kansai", "chugoku", "shikoku", "kyushu".	エリアのパラメータが不正な場合
4	404	NO_DATA_ON_DATE	No data found for the specified date.	指定された日付に対するデータが存在しない場合
5	404	NO_AREA_DATA_ON_DATE	No data found for the specified area "{area}".	指定されたエリアに対するデータが存在しない場合

実装

実装はFastAPIを使用した。

コード

main.py

from fastapi import FastAPI, HTTPException
import datetime
from model import get_data_from_firestore

app = FastAPI()

# APIのエンドポイント
@app.get("/api/electricity_market_price")
def get_electricity_market_price(date: str = None, area: str = None):
    # パラメータのバリデーション
    validate_params(date, area)

    # Firestoreからデータの取得
    data = get_data_from_firestore(date, area)

    # 正常なレスポンスの返却
    return data

# パラメータバリデーション
def validate_params(date: str, area: str) -> None:
    # 必須パラメータdateが指定されていない場合
    if date is None:
        error_type = "MISSING_REQUIRED_PARAMETER_DATE"
        error_message = "Date parameter is missing."
        raise HTTPException(
            status_code=400,
            detail={'type':error_type, 'message':error_message}
        )
    
    # パラメータdateのフォーマットがyyyy-mm-ddか確認
    try:
        datetime.datetime.strptime(date, '%Y-%m-%d')
    except ValueError:
        error_type = "INVALID_DATE_FORMAT"
        error_message = "Date parameter must be in the format of yyyy-mm-dd."
        raise HTTPException(
            status_code=400,
            detail={'type':error_type, 'message':error_message}
        )
    
    # パラメータareaが指定されている場合は、指定されたエリアが許容されているか確認
    if area:
        area_list = area.split(',')
        # 許容するエリア
        valid_areas = ['system', 'hokkaido', 'tohoku', 'tokyo', 'hokuriku', 'chubu', 'kansai', 'chugoku', 'shikoku', 'kyushu']
        if not all(area in valid_areas for area in area_list):
            error_type = "INVALID_AREA_PARAMETER"
            error_message = "Invalid area parameter. Allowed values are 'system', 'hokkaido', 'tohoku', 'tokyo', 'hokuriku', 'chubu', 'kansai', 'chugoku', 'shikoku', 'kyushu'"
            raise HTTPException(
                status_code=400,
                detail={'type':error_type, 'message':error_message}
            )

model.py

from google.cloud import firestore
import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
from fastapi import HTTPException

# Firestoreの認証
cred = credentials.Certificate('******************.json')
firebase_admin.initialize_app(cred)

# Cloud Firestoreに接続
db = firestore.client()

def get_data_from_firestore(date: str, area: str) -> dict:
    """
    Firestoreから指定された日付とエリアの電力市場価格データを取得する関数。

    Parameters:
    -----------
    date : str
        取得する電力市場価格データの日付（yyyy-mm-dd形式の文字列）。
    area : str
        取得する電力市場価格データのエリア。指定しない場合はNone。

    Returns:
    --------
    data : Dict[str, Dict[str, float]]
        指定された日付とエリアの電力市場価格データが格納された辞書型オブジェクト。
        キーはエリア名、値は時刻とエリアの価格が格納された辞書型オブジェクト。
    """
    # Firestoreからデータの取得
    doc_ref = db.collection('electricity_market_price').document(date)
    doc = doc_ref.get()

    # 指定した日付のデータが取得できなかった場合
    if not doc.exists:
        error_type = f"NO_DATA_ON_DATE"
        error_message = f"No data found for the specified date.'{date}'."
        raise HTTPException(
            status_code=404,
            detail={'type':error_type, 'message':error_message}
        )

    # FirestoreのDocumentSnapshotクラスをdict型に変換
    data = doc.to_dict()

    # エリアが指定されている場合、日付で絞り込んだデータをエリアで絞り込む
    if area:
        area_list = area.split(',')
        selected_data = {}
        for request_area in area_list:
            # 指定した日付のデータに指定したエリアのキーが存在しない場合
            if request_area not in data:
                error_type = f"NO_AREA_DATA_ON_DATE"
                error_message = f"No data found for the specified area {request_area} on date {date}."
                raise HTTPException(
                    status_code=404,
                    detail={'type':error_type, 'message':error_message}
                )
            selected_data[request_area] = data[request_area]
        data = selected_data

    return data

各処理の説明は下記に別途記載する。

パラメータのバリデーション

パラメータ`date`の存在確認

パラメータdateが指定されていない場合は、処理を中断しエラーNo.1を返す。

if date is None:
    error_type = "MISSING_REQUIRED_PARAMETER_DATE"
    error_message = "Date parameter is missing."
    raise HTTPException(
        status_code=400,
        detail={'type':error_type, 'message':error_message}
    )

パラメータ`date`のフォーマット確認

URLエンコードされたパラメータdateをデコードし、許容するフォーマットyyyy-mm-dd通りか確認する。
異なる場合は、処理を中断しエラーNo.2を返す。

# パラメータdateのフォーマットがyyyy-mm-ddか確認
try:
    datetime.datetime.strptime(date, '%Y-%m-%d')
except ValueError:
    error_type = "INVALID_DATE_FORMAT"
    error_message = "Date parameter must be in the format of yyyy-mm-dd."
    raise HTTPException(
        status_code=400,
        detail={'type':error_type, 'message':error_message}
    )

パラメータ`area`の内容確認

パラメータareaに指定のエリア以外が含まれていないか確認する。含まれている場合は、処理を中断しエラーNo.3を返す。
指定のエリアは以下の通りとする。

system
hokkaido
tohoku
tokyo
hokuriku
chubu
kansai
chugoku
shikoku
kyushu

# 許容するエリア
valid_areas = ['system', 'hokkaido', 'tohoku', 'tokyo', 'hokuriku', 'chubu', 'kansai', 'chugoku', 'shikoku', 'kyushu']
if not all(area in valid_areas for area in area_list):
    error_type = "INVALID_AREA_PARAMETER"
    error_message = "Invalid area parameter. Allowed values are 'system', 'hokkaido', 'tohoku', 'tokyo', 'hokuriku', 'chubu', 'kansai', 'chugoku', 'shikoku', 'kyushu'"
    raise HTTPException(
        status_code=400,
        detail={'type':error_type, 'message':error_message}
    )

データ取得

日付の絞り込み

パラメータdateで指定された日付に該当するデータを、Firestoreから取得する。
なお取得したデータは、FirestoreのDocumentSnapshotクラスとなっている。

データが存在しない場合は、処理を中断しエラーNo.4を返す。

# 指定されたGCPプロジェクトのFirestoreのクライアントオブジェクトを作成
db = firestore.Client(project=PROJECT_ID)

# 'electricity_market_price/"date"'の参照を取得
doc_ref = db.collection('electricity_market_price').document(date)

# 参照したドキュメントの内容を取得
doc = doc_ref.get()

# 指定した日付のデータが取得できなかった場合
if not doc.exists:
    error_type = f"NO_DATA_ON_DATE"
    error_message = f"No data found for the specified date.'{date}'."
    raise HTTPException(
        status_code=404,
        detail={'type':error_type, 'message':error_message}
    )

エリアの絞り込み

パラメータareaが指定されている場合は、
指定されたエリアのデータのみを{エリア名:エリアプライス}, {エリア名:エリアプライス},...の形式で抽出する。
指定されていない場合は、絞り込みは行わない。

日付で絞り込んだデータに、指定したエリアのキーが存在しない場合、処理を中断しエラーNo.5を返す。

# FirestoreのDocumentSnapshotクラスをdict型に変換
data = doc.to_dict()

# エリアが指定されている場合、日付で絞り込んだデータをエリアで絞り込む
if area:
    area_list = area.split(',')
    selected_data = {}
    for request_area in area_list:
        # 指定した日付のデータに指定したエリアのキーが存在しない場合
        if request_area not in data:
            error_type = f"NO_AREA_DATA_ON_DATE"
            error_message = f"No data found for the specified area {request_area} on date {date}."
            raise HTTPException(
                status_code=404,
                detail={'type':error_type, 'message':error_message}
            )
        # エリアをキーとするデータを追加
        selected_data[request_area] = data[request_area]
    data = selected_data

正常なレスポンスの返却

正常にデータが取得できた場合、データを返す。
FastAPIの場合、次のようにレスポンスを返すだけで、自動的にステータスコード200が付与される。

return data

デプロイ

Cloud Functionsにデプロイする。
Cloud Functionsには2つの世代があるが、今回は起動時間が短く、高速でスケーラブルな第二世代を採用する。

今回Web API開発にあたり、PythonのwebフレームワークであるFastAPIを使用したが、残念ながらそのままではCloud Functions, AWS lambdaなどのサーバーレスサービスでは稼働することはできない。

これは、FastAPIはASGI（Asynchronous Server Gateway Interface）アプリケーションフレームワークなのに対し、Cloud Functions, AWS lambdaなどのサーバーレスサービスはWSGI（Web Server Gateway Interface）アプリケーションフレームワークをサポートしており、これらのインターフェースには互換性がないためだ。

そこでPythoonモジュール「Agraffe」を使用することで、Cloud function, AWS lambdaなどのサーバーレスサービスにASGIサーバを構築できるようにする。

修正点

以上を踏まえた修正点は以下の通り

エンドポイントを追加する。
```
entry_point = Agraffe.entry_point(app)
```
Firestoreの認証をIAMロールで自動的に行うため、既存の認証処理を削除する。

ファイル構成

ファイル構成は以下の通り。

project/
├── main.py
├── model.py
└── requirements.txt

main.py: メインのアプリケーションコードが含まれるファイル
model.py: データモデルに関するコードが含まれるファイル
requirements.txt: アプリケーションの依存パッケージを定義するファイル

コード

main.py

from agraffe import Agraffe
from fastapi import FastAPI, HTTPException
import datetime
from model import get_data_from_firestore

app = FastAPI()

@app.get("/api/electricity_market_price")
async def get_electricity_market_price(date: str = None, area: str = None):
    # # パラメータのバリデーション
    validate_params(date, area)

    # # Firestoreからデータの取得
    data = get_data_from_firestore(date, area)

    # 正常なレスポンスの返却
    return data

# '/api/electricity_market_price'以外は404を返すように設定
@app.get("/{path}")
async def invalid_path(path: str):
    raise HTTPException(status_code=404, detail="Not Found")

# パラメータバリデーション
def validate_params(date: str, area: str) -> None:
    # 必須パラメータdateが指定されていない場合
    if date is None:
        error_type = f"MISSING_REQUIRED_PARAMETER_DATE"
        error_message = f"Date parameter is missing."
        raise HTTPException(
            status_code=400,
            detail={'type':error_type, 'message':error_message}
        )
    
    # パラメータdateのフォーマットがyyyy-mm-ddか確認
    try:
        datetime.datetime.strptime(date, '%Y-%m-%d')
    except ValueError:
        error_type = f"INVALID_DATE_FORMAT"
        error_message = f"Date parameter must be in the format of yyyy-mm-dd."
        raise HTTPException(
            status_code=400,
            detail={'type':error_type, 'message':error_message}
        )
    
    # パラメータareaが指定されている場合は、指定されたエリアが許容されているか確認
    if area:
        area_list = area.split(',')
        # 許容するエリア
        valid_areas = ['system', 'hokkaido', 'tohoku', 'tokyo', 'hokuriku', 'chubu', 'kansai', 'chugoku', 'shikoku', 'kyushu']
        if not all(area in valid_areas for area in area_list):
            error_type = f"INVALID_AREA_PARAMETER"
            error_message = f"Invalid area parameter. Allowed values are 'system', 'hokkaido', 'tohoku', 'tokyo', 'hokuriku', 'chubu', 'kansai', 'chugoku', 'shikoku', 'kyushu'"
            raise HTTPException(
                status_code=400,
                detail={'type':error_type, 'message':error_message}
            )
entry_point = Agraffe.entry_point(app)

model.py

from google.cloud import firestore
from fastapi import HTTPException

db = firestore.Client()

def get_data_from_firestore(date: str, area: str) -> dict:
    """
    Firestoreから指定された日付とエリアの電力市場価格データを取得する関数。

    Parameters:
    -----------
    date : str
        取得する電力市場価格データの日付（yyyy-mm-dd形式の文字列）。
    area : str
        取得する電力市場価格データのエリア。指定しない場合はNone。

    Returns:
    --------
    data : Dict[str, Dict[str, float]]
        指定された日付とエリアの電力市場価格データが格納された辞書型オブジェクト。
        キーはエリア名、値は時刻とエリアの価格が格納された辞書型オブジェクト。
    """
    # Firestoreからデータの取得
    doc_ref = db.collection('electricity_market_price').document(date)
    doc = doc_ref.get()

    # 指定した日付のデータが取得できなかった場合
    if not doc.exists:
        error_type = f"NO_DATA_ON_DATE"
        error_message = f"No data found for the specified date.'{date}'."
        raise HTTPException(
            status_code=404,
            detail={'type':error_type, 'message':error_message}
        )

    # FirestoreのDocumentSnapshotクラスをdict型に変換
    data = doc.to_dict()

    # エリアが指定されている場合、日付で絞り込んだデータをエリアで絞り込む
    if area:
        area_list = area.split(',')
        selected_data = {}
        for request_area in area_list:
            # 指定した日付のデータに指定したエリアのキーが存在しない場合
            if request_area not in data:
                error_type = f"NO_AREA_DATA_ON_DATE"
                error_message = f"No data found for the specified area '{request_area}' on date '{date}'."
                raise HTTPException(
                    status_code=404,
                    detail={'type':error_type, 'message':error_message}
                )
            # エリアをキーとするデータを追加
            selected_data[request_area] = data[request_area]
        data = selected_data

    return data

requirements.txt

agraffe==0.7.0
fastapi==0.95.1
datetime==4.3
google-cloud-firestore==2.11.0

Cloud Functionsの構成

以下構成で構築する。

項目	設定内容
関数名	get_electricity_price
環境	第二世代
リージョン	us-central1
トリガータイプ	http
認証	未認証の呼び出しを許可
メモリ	256MB
タイムアウト	60sec
ランタイム	Python3.11
エントリーポイント	entry_point(Agraffe)

※認証は後々変更する予定

デプロイコマンド

デプロイするファイルが含まれるディレクトリで以下コマンドを実行する。

 gcloud functions deploy get_electricity_price --gen2 --region=us-central1B --trigger-http --allow-unauthenticated --memory=256M --timeout 60s --runtime=python311 --entry-point=entry_point

結果

正常時レスポンス
例) url=https://******/api/electricity_market_price?date=2023-04-01&area=tokyo
エラーレスポンス
例) url=https://******/api/electricity_market_price?date=2023-04-01&area=tokyo,kyushuuuu

おわりに

今後の展開

Web APIで当日分、時間によっては（基本10時以降）翌日分のエリアプライスを取得できる環境が構築できた。
今後は、取得したデータをもとにアプリやマイコンなどで、より情報取得の障壁を下げて節電に取り組むことや、自動的に消費電力の大きい家電の電源をOFFにするなどの取り組みに繋げたい。

日本卸電力取引所のエリアプライスを返すAPIをFastAPIを使用してCloud Functionsにデプロイした話