FastAPIのlifespanで起動時discover→バックグラウンド定期ポーリングを回す(python-kasa / Tapo P110M)

概要

スマートプラグ Tapo P110M の消費電力を、FastAPI 製のダッシュボードで監視する構成です。python-kasa の API はすべて async で、しかも discover は数秒かかります。そこで アプリ起動時に 1 回だけ discover して接続を使い回し、以降はバックグラウンドタスクで一定間隔ポーリングして SQLite に記録 します。この記事では FastAPI の lifespan の書き方と、通信断からデバイス単位で復旧する部分だけ抜き出します。

つまずいたポイント

最初はエンドポイントの中で毎回 Discover.discover() を呼んでいましたが、1 リクエストに数秒かかるうえ、デバイスにも余計な負荷がかかります。

接続は「起動時に 1 回」に寄せたいのですが、FastAPI 単体だと「起動時に 1 回だけ実行して、その結果をエンドポイント全体で共有する」置き場所に迷います。ここで使うのが lifespan(旧 @app.on_event("startup") の後継)です。起動時に discover → asyncio.create_task() でポーリングループを開始し、終了時にタスクをキャンセルして切断します。

コア1: lifespanで1回だけdiscoverし、ポーリングタスクを起動する

import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI

from tapo_devices import discover_with_retry, disconnect_all

devices: dict = {}          # 検出したデバイスを全体で共有する

@asynccontextmanager
async def lifespan(app: FastAPI):
    global devices
    devices = await discover_with_retry()      # 起動時に1回だけ
    for dev in devices.values():
        await dev.update()
    task = asyncio.create_task(poll_loop())    # バックグラウンドで回し続ける
    try:
        yield                                  # ここでアプリが動く
    finally:
        task.cancel()                          # 終了時に後片付け
        await disconnect_all(devices)

app = FastAPI(lifespan=lifespan)

エンドポイント側は devices や、ポーリングが更新する latest_status を参照するだけで済みます。

コア2: 通信失敗をデバイス単位で数えて、閾値で再discoverする

Wi-Fi 環境では dev.update() がたまに失敗します。全体を作り直すと重いので、デバイスごとに連続失敗回数を数え、閾値(既定 2 回)を超えたそのデバイスだけ discover し直します。失敗中も直前の値に online: False を上書きして残し、画面の OFFLINE 表示に使います。

POLL_INTERVAL_SEC = 15
REDISCOVER_AFTER_FAILURES = 2

latest_status: dict = {}
consecutive_failures: dict = {}

async def poll_loop():
    conn = init_db()
    while True:
        for name, dev in list(devices.items()):
            try:
                await dev.update()
                log_reading(conn, name, dev)
                latest_status[name] = status_of(dev, online=True)
                consecutive_failures[name] = 0
            except Exception as exc:
                consecutive_failures[name] = consecutive_failures.get(name, 0) + 1
                latest_status[name] = {**latest_status.get(name, {}),
                                       "online": False, "error": str(exc)}
                if consecutive_failures[name] >= REDISCOVER_AFTER_FAILURES:
                    await rediscover_one(name)   # この1台だけ discover し直す
        await asyncio.sleep(POLL_INTERVAL_SEC)

なお Tapo P110M の Energy モジュールは consumption_today / consumption_this_month は取れましたが、consumption_total(生涯積算)は実機で常に None でした。

完全なソースコードと手順はブログにまとめています

power_readings テーブルの設計、dev.modules["Energy"] から取れる値・取れない値の一覧、/api/status・/api/history・ON/OFF API の実装、バニラ JS のフロント、ポーリング間隔を環境変数にした理由、常駐化を見送った判断まで、以下の記事に書いています。