blastengine APIのHTTPエラーハンドリングを解説

メール配信API「blastengine」を利用する際、適切なエラーハンドリングは安定したシステム運用に欠かせません。本記事では、blastengine APIのステータス構造を理解し、実践的なエラーハンドリング方法を解説します。

blastengine APIのステータス構造を理解する

blastengine APIでは、2段階のステータス構造を採用しています。この構造を正しく理解することが、効果的なエラーハンドリングの第一歩です。

第1段階：HTTPステータスコード（API呼び出しレベル）

API呼び出し時に返されるHTTPステータスコードは、リクエストの成否を示します。

• 200系：リクエストが正常に処理された
• 400系：クライアント側のエラー（リクエスト内容の不備、認証失敗など）
• 500系：サーバー側のエラー（内部エラー、高負荷など）

第2段階：配信ジョブステータス（メール配信レベル）

API呼び出しが成功（200系）した後、実際のメール配信状況は配信ステータスで管理されます。

ステータス	説明
EDIT	編集中
IMPORTING	配信アドレス一括インポート中
RESERVE	配信予約済
WAIT	配信待ち
SENDING	配信中
SENT	配信成功
FAILED	配信失敗

重要なポイント：API呼び出しが成功（HTTP 200）しても、メール配信自体が失敗する可能性があります。そのため、両方のステータスを適切に監視する必要があります。

よく使うHTTPステータス一覧と意味

blastengine APIで返される主要なHTTPステータスコードと、その対応方法を一覧にまとめました。

ステータスコード対応表

コード	種別	説明	推奨される対応
200	OK	リクエスト成功	正常処理
201	Created	リソース作成成功	正常処理（配信ID等を保存）
400	Bad Request	リクエスト値が不正	リクエスト内容を修正
401	Unauthorized	認証エラー	BearerTokenを確認
403	Forbidden	API利用権限なし	契約・権限を確認
404	Not Found	指定リソースが存在しない	リソースIDを確認
405	Method Not Allowed	HTTPメソッドが未サポート	HTTPメソッドを確認
415	Unsupported Media Type	リクエスト形式が未サポート	Content-Typeヘッダを確認
423	Locked	アカウントロック	サポートに問い合わせ
429	Too Many Requests	Rate Limit超過	待機後に再送（後述）
500	Internal Server Error	サーバエラー	リトライ処理を実装
502	Bad Gateway	web通信の高負荷	リトライ処理を実装

Rate Limitについて

blastengine APIのRate Limitは500req/m（1分間に500リクエスト）です。これを超えると429エラーが返されます。

429エラー発生時は、以下のレスポンスヘッダーが返されます：

• X-Rate-Limit-Remaining：アクセス可能な残り回数
• X-Rate-Limit-Retry-After-Seconds：アクセス回数がリセットされるまでの時間（秒）

PythonでのHTTPステータスハンドリング実装例

実際のコードでHTTPステータスを適切にハンドリングする方法を見ていきましょう。

本記事では、blastengine公式SDKを使用した実装例を紹介します。SDKを使うことで、BearerTokenの生成やAPIリクエストの構築が自動化され、開発効率が大幅に向上します。

blastengine SDK版のクライアント実装

公式SDKを使いつつ、HTTPレスポンスの詳細なエラーハンドリングを実装したクライアントクラスです。

import requests
import time
import hashlib
import base64
import logging
from typing import Optional, Dict, Any, Callable
from blastengine.Client import Blastengine
from blastengine.Transaction import Transaction

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class RateLimitError(Exception):
    """Rate Limit超過エラー"""
    def __init__(self, response: requests.Response):
        self.retry_after = int(response.headers.get('X-Rate-Limit-Retry-After-Seconds', 60))
        self.remaining = int(response.headers.get('X-Rate-Limit-Remaining', 0))
        super().__init__(f"Rate Limit超過。{self.retry_after}秒後に再試行してください")


class BlastengineClient:
    """
    blastengine SDK版クライアント

    公式SDKを使用しつつ、HTTPレスポンスの詳細なエラーハンドリングを提供
    """

    def __init__(self, login_id: str, api_key: str):
        """クライアントを初期化"""
        self.sdk_client = Blastengine(login_id, api_key)
        self.base_url = "https://app.engn.jp/api/v1"
        self.bearer_token = self.sdk_client.token
        self.headers = {
            "Authorization": f"Bearer {self.bearer_token}",
            "Content-Type": "application/json",
            "Accept-Language": "ja-JP"
        }
        self.last_response: Optional[requests.Response] = None

    def _handle_response(self, response: requests.Response) -> Dict[Any, Any]:
        """レスポンスを処理してエラーハンドリングを実行"""
        self.last_response = response

        # 成功レスポンス
        if response.status_code in [200, 201]:
            logger.info(f"✓ HTTPステータス: {response.status_code}")
            return response.json()

        # クライアントエラー
        if 400 <= response.status_code < 500:
            if response.status_code == 400:
                raise ValueError(f"リクエスト不正: {response.text}")
            elif response.status_code == 401:
                raise PermissionError("認証エラー: BearerTokenを確認してください")
            elif response.status_code == 429:
                raise RateLimitError(response)
            else:
                raise Exception(f"クライアントエラー ({response.status_code}): {response.text}")

        # サーバーエラー
        if response.status_code >= 500:
            raise Exception(f"サーバーエラー ({response.status_code}): {response.text}")

        return response.json()

    def send_transaction_mail(
        self,
        from_email: str,
        to_email: str,
        subject: str,
        text_part: str,
        from_name: str = "",
        html_part: Optional[str] = None
    ) -> Dict[Any, Any]:
        """
        トランザクションメールを送信

        SDKを使用しつつ、HTTPレスポンスをキャプチャして
        エラーハンドリングを実行します
        """
        mail = Transaction()
        mail.subject(subject)
        mail.from_address(from_email, from_name)
        mail.to(to_email)
        mail.text_part(text_part)
        if html_part:
            mail.html_part(html_part)

        # HTTPレスポンスをキャプチャ
        original_post = requests.post
        captured_response = None

        def intercepted_post(*args, **kwargs):
            nonlocal captured_response
            captured_response = original_post(*args, **kwargs)
            return captured_response

        requests.post = intercepted_post

        try:
            delivery_id = mail.send()

            if captured_response:
                result = self._handle_response(captured_response)
                return result
            else:
                return {"delivery_id": delivery_id}
        finally:
            requests.post = original_post


def retry_with_backoff(
    func: Callable,
    max_retries: int = 3,
    backoff_factor: float = 2.0
) -> Any:
    """指数バックオフでリトライを実行"""
    for attempt in range(max_retries):
        try:
            return func()

        except RateLimitError as e:
            logger.warning(f"Rate Limit超過。{e.retry_after}秒待機します...")
            logger.info(f"残りリクエスト数: {e.remaining}")
            time.sleep(e.retry_after)

            if attempt == max_retries - 1:
                logger.error(f"最大リトライ回数({max_retries})に達しました")
                raise

        except Exception as e:
            if attempt == max_retries - 1:
                logger.error(f"最大リトライ回数({max_retries})に達しました: {e}")
                raise

            wait_time = backoff_factor ** attempt
            logger.warning(f"エラー発生。{wait_time}秒後にリトライします (試行{attempt + 1}/{max_retries})")
            time.sleep(wait_time)

使用例

# クライアント初期化
client = BlastengineClient("your_login_id", "your_api_key")

def send_email():
    """メール送信"""
    return client.send_transaction_mail(
        from_email="[e-mail]",
        from_name="送信者名",
        to_email="[e-mail]",
        subject="テストメール",
        text_part="これはテストメールです。"
    )

# Rate Limit対応でリトライ実行
try:
    result = retry_with_backoff(send_email)
    logger.info(f"配信ID: {result.get('delivery_id')} で登録成功")

    # HTTPレスポンスの詳細を確認
    if client.last_response:
        print(f"ステータスコード: {client.last_response.status_code}")
        print(f"Rate Limit残り: {client.last_response.headers.get('X-Rate-Limit-Remaining')}")

except Exception as e:
    logger.error(f"メール送信に失敗しました: {e}")

まとめ

blastengine APIを安定運用するためのエラーハンドリングのポイントを振り返りましょう。

1. 2段階のステータスを理解する

   • HTTPステータスコード：API呼び出しの成否
   • 配信ステータス：メール配信の状況
   • 両方を適切に監視することが重要

2. 適切なリトライ戦略を実装する

   • 429（Rate Limit）：Retry-Afterヘッダーに従って待機
   • 500系（サーバーエラー）：指数バックオフでリトライ
   • 400系（クライアントエラー）：即時アラート、リトライ不要

blastengine APIのエラーハンドリングを適切に実装することで、メール配信システムの信頼性と運用効率が大幅に向上します。本記事の実装例は19個のユニットテストで全てのエラーハンドリングパスを検証済みですので、安心してプロジェクトに組み込んでください。