blastengine 利用状況APIでメール配信コストを可視化する

はじめに

メール配信サービスを運用していると「今月あと何通送れるのか」「プランの上限に近づいていないか」が気になります。blastengine APIには利用状況を取得するusagesエンドポイントが用意されており、配信量の監視やアラート通知を自動化できます。

本記事では、利用状況APIを使った配信コストの可視化と、上限アラートの実装方法を解説します。

usages APIの概要

利用状況APIは3つのエンドポイントで構成されています。

エンドポイント	メソッド	用途
/usages	GET	月別の利用状況一覧を取得
/usages/{month}	GET	指定月の利用状況詳細を取得
/usages/latest	GET	最新の利用状況を取得

レスポンスの構造

{
  "month": 202111,
  "current": 50000,
  "remaining": 250000,
  "updated_time": "2021-11-01T01:00:00+09:00",
  "plan_id": "be-plan-300000"
}

フィールド	内容
month	対象年月（YYYYMM形式）
current	当月の配信数
remaining	残りの配信可能数
updated_time	最終更新日時
plan_id	契約プランID

current + remaining でプランの月間上限が算出できます。

環境準備

pip install requests

import requests
from datetime import datetime

API_BASE = "https://app.engn.jp/api/v1"
BEARER_TOKEN = "your_bearer_token"

HEADERS = {
    "Authorization": f"Bearer {BEARER_TOKEN}",
    "Content-Type": "application/json",
}

実装1：最新の利用状況を取得する

最もシンプルな使い方です。現在の配信数と残数をワンコールで取得します。

def get_latest_usage():
    """最新の利用状況を取得"""
    url = f"{API_BASE}/usages/latest"
    response = requests.get(url, headers=HEADERS)

    if response.status_code != 200:
        print(f"取得失敗: {response.status_code}")
        return None

    data = response.json()
    total = data["current"] + data["remaining"]
    usage_rate = (data["current"] / total) * 100

    print(f"対象月: {data['month']}")
    print(f"プラン: {data['plan_id']}")
    print(f"配信数: {data['current']:,} / {total:,}")
    print(f"残数:   {data['remaining']:,}")
    print(f"使用率: {usage_rate:.1f}%")
    return data


get_latest_usage()

出力例：

対象月: 202603
プラン: be-plan-300000
配信数: 185,000 / 300,000
残数:   115,000
使用率: 61.7%

実装2：月別の利用推移を取得する

/usagesエンドポイントのmonth_agoパラメータで過去の利用状況をまとめて取得できます。

def get_usage_history(months=6):
    """過去N月分の利用状況を取得"""
    url = f"{API_BASE}/usages"
    params = {"month_ago": months}
    response = requests.get(url, headers=HEADERS, params=params)

    if response.status_code != 200:
        print(f"取得失敗: {response.status_code}")
        return None

    data = response.json()["data"]

    print(f"{'年月':>8} | {'配信数':>10} | {'残数':>10} | {'使用率':>6}")
    print("-" * 50)
    for item in data:
        total = item["current"] + item["remaining"]
        rate = (item["current"] / total) * 100 if total > 0 else 0
        print(
            f"{item['month']:>8} | {item['current']:>10,} | "
            f"{item['remaining']:>10,} | {rate:>5.1f}%"
        )

    return data


get_usage_history(months=6)

出力例：

    年月 |       配信数 |         残数 |  使用率
--------------------------------------------------
  202603 |    185,000 |    115,000 |  61.7%
  202602 |    220,000 |     80,000 |  73.3%
  202601 |    195,000 |    105,000 |  65.0%
  202512 |    210,000 |     90,000 |  70.0%
  202511 |    180,000 |    120,000 |  60.0%
  202510 |    175,000 |    125,000 |  58.3%

実装3：使用率アラート通知

プランの上限に近づいたら通知する仕組みを作ります。ここではSlack Webhook連携の例を示します。

import json

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/XXX/YYY/ZZZ"
ALERT_THRESHOLD = 80  # 使用率80%でアラート


def check_and_alert(threshold=ALERT_THRESHOLD):
    """使用率が閾値を超えたらSlackに通知"""
    url = f"{API_BASE}/usages/latest"
    response = requests.get(url, headers=HEADERS)

    if response.status_code != 200:
        print(f"取得失敗: {response.status_code}")
        return

    data = response.json()
    total = data["current"] + data["remaining"]
    usage_rate = (data["current"] / total) * 100

    if usage_rate >= threshold:
        message = (
            f":warning: *blastengine 配信数アラート*\n"
            f"対象月: {data['month']}\n"
            f"プラン: {data['plan_id']}\n"
            f"配信数: {data['current']:,} / {total:,}\n"
            f"使用率: {usage_rate:.1f}%\n"
            f"残数: {data['remaining']:,}"
        )
        slack_payload = {"text": message}
        requests.post(
            SLACK_WEBHOOK_URL,
            data=json.dumps(slack_payload),
            headers={"Content-Type": "application/json"},
        )
        print(f"アラート送信済み（使用率: {usage_rate:.1f}%）")
    else:
        print(f"正常範囲内（使用率: {usage_rate:.1f}%）")


check_and_alert()

cronで定期実行

このスクリプトを日次で実行すれば、上限超過を未然に防げます。

# 毎日9時に実行
0 9 * * * /usr/bin/python3 /path/to/check_usage.py

実装4：月別利用推移のCSV出力

レポート用にCSVとして出力する実装です。

import csv


def export_usage_csv(months=12, output_path="usage_report.csv"):
    """利用状況をCSVに出力"""
    url = f"{API_BASE}/usages"
    params = {"month_ago": months}
    response = requests.get(url, headers=HEADERS, params=params)

    if response.status_code != 200:
        print(f"取得失敗: {response.status_code}")
        return

    data = response.json()["data"]

    with open(output_path, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.writer(f)
        writer.writerow(["年月", "配信数", "残数", "プラン上限", "使用率(%)"])

        for item in data:
            total = item["current"] + item["remaining"]
            rate = round((item["current"] / total) * 100, 1) if total > 0 else 0
            writer.writerow([
                item["month"],
                item["current"],
                item["remaining"],
                total,
                rate,
            ])

    print(f"CSV出力完了: {output_path}")


export_usage_csv(months=12)

実装5：指定月の詳細取得

特定月の利用状況をピンポイントで確認する場合は /usages/{month} を使います。

def get_monthly_usage(year, month):
    """指定月の利用状況を取得"""
    month_str = f"{year}{month:02d}"
    url = f"{API_BASE}/usages/{month_str}"
    response = requests.get(url, headers=HEADERS)

    if response.status_code != 200:
        print(f"取得失敗: {response.status_code}")
        return None

    data = response.json()
    total = data["current"] + data["remaining"]

    print(f"年月: {data['month']}")
    print(f"プラン: {data['plan_id']}")
    print(f"配信数: {data['current']:,} / {total:,}")
    print(f"最終更新: {data['updated_time']}")
    return data


# 2026年3月の利用状況を確認
get_monthly_usage(2026, 3)

運用のポイント

観点	推奨
監視頻度	日次（cronなど）で/usages/latestをチェック
アラート閾値	80%で警告、95%で緊急通知など段階的に設定
履歴保存	月次で/usagesの結果をCSV/DBに蓄積して傾向分析
プラン見直し	3ヶ月連続で使用率70%超なら上位プラン検討

おわりに

blastengineの利用状況APIは3エンドポイントとシンプルですが、配信コストの可視化・アラート通知・レポート出力まで十分カバーできます。

「気づいたら上限に達していて配信できなかった」という事故を防ぐためにも、日次の監視スクリプトを仕込んでおくことをおすすめします。