blastengine API 共通仕様まとめ（Python実装例付き）

API共通仕様の整理

blastengine APIは、メール配信サービスを提供するREST APIです。以下の共通仕様に従って実装します。

基本仕様

項目	説明
APIバージョン	1.19.0
ベースURL	https://app.engn.jp/api/v1/
プロトコル	HTTPS
文字コード	UTF-8
データ形式	JSON (一部multipart/form-data)
Rate Limit	500リクエスト/分

認証方式

blastengine APIはBearer Token認証を使用します。トークンの生成方法は以下の通りです。

1. ログインID + APIキーをSHA256でハッシュ化
2. 大文字を小文字に変換
3. Base64エンコード

YOUR_LOGIN_ID=ログインID
YOUR_API_KEY=APIキー
YOUR_BEARER_TOKEN=$(echo -n ${YOUR_LOGIN_ID}${YOUR_API_KEY} | shasum -a 256 | awk '{print $1}')
YOUR_BEARER_TOKEN=$(echo -n ${YOUR_BEARER_TOKEN} | tr A-Z a-z)
YOUR_BEARER_TOKEN=$(echo -n ${YOUR_BEARER_TOKEN} | base64 | tr -d "\n")
echo ${YOUR_BEARER_TOKEN}

共通リクエストヘッダ（Python例）

APIリクエストには以下のヘッダーが必要です。

必須ヘッダー

import requests
import os

class BlastEngineClient:
    """blastengine API クライアント"""

    def __init__(self, bearer_token: str):
        self.base_url = "https://app.engn.jp/api/v1"
        self.bearer_token = bearer_token

    def _get_headers(self, content_type: str = "application/json") -> dict:
        """共通リクエストヘッダーを取得"""
        return {
            "Authorization": f"Bearer {self.bearer_token}",
            "Content-Type": content_type,
            "Accept-Language": "ja-JP"  # または en-US
        }

# 使用例
bearer_token = os.environ.get("BLASTENGINE_BEARER_TOKEN")
client = BlastEngineClient(bearer_token)

ヘッダー詳細

ヘッダー	説明	例
Authorization	Bearer Token認証情報（必須）	Bearer {token}
Content-Type	リクエストボディの形式	application/json
Accept-Language	ロケール情報	ja-JP / en-US

最小リクエスト例（Requests）

def get_delivery_list(client: BlastEngineClient, page: int = 1, size: int = 100):
    """配信一覧を取得する最小例"""
    url = f"{client.base_url}/deliveries/all"
    headers = client._get_headers()
    params = {
        "page": page,
        "size": size,
        "sort": "delivery_time:desc"
    }

    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()  # HTTPエラーの場合は例外を発生

    return response.json()

# 実行例
try:
    result = get_delivery_list(client, page=1, size=10)
    print(f"取得件数: {len(result.get('data', []))}")
except requests.exceptions.HTTPError as e:
    print(f"HTTPエラー: {e}")
except requests.exceptions.RequestException as e:
    print(f"リクエストエラー: {e}")

タイムアウト・リトライ設定

本番環境では、タイムアウトとリトライ処理を実装することを推奨します。

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

class BlastEngineClient:
    """改良版 blastengine API クライアント（タイムアウト・リトライ対応）"""

    def __init__(self, bearer_token: str, timeout: int = 30, max_retries: int = 3):
        self.base_url = "https://app.engn.jp/api/v1"
        self.bearer_token = bearer_token
        self.timeout = timeout

        # セッションの設定
        self.session = requests.Session()

        # リトライ戦略の設定
        retry_strategy = Retry(
            total=max_retries,
            backoff_factor=1,  # 1秒, 2秒, 4秒...と待機時間を増やす
            status_forcelist=[429, 500, 502, 503, 504],  # リトライ対象のステータスコード
            allowed_methods=["GET", "POST", "PUT", "PATCH", "DELETE"]
        )

        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)

    def _get_headers(self, content_type: str = "application/json") -> dict:
        """共通リクエストヘッダーを取得"""
        return {
            "Authorization": f"Bearer {self.bearer_token}",
            "Content-Type": content_type,
            "Accept-Language": "ja-JP"
        }

    def get(self, endpoint: str, params: dict = None) -> dict:
        """GETリクエスト"""
        url = f"{self.base_url}/{endpoint}"
        headers = self._get_headers()

        response = self.session.get(
            url,
            headers=headers,
            params=params,
            timeout=self.timeout
        )
        response.raise_for_status()
        return response.json()

    def post(self, endpoint: str, data: dict = None) -> dict:
        """POSTリクエスト"""
        url = f"{self.base_url}/{endpoint}"
        headers = self._get_headers()

        response = self.session.post(
            url,
            headers=headers,
            json=data,
            timeout=self.timeout
        )
        response.raise_for_status()
        return response.json()

# 使用例
client = BlastEngineClient(
    bearer_token=os.environ.get("BLASTENGINE_BEARER_TOKEN"),
    timeout=30,
    max_retries=3
)

レスポンスヘッダ・共通レスポンス仕様

レスポンスヘッダー

blastengine APIから返される共通レスポンスヘッダーは以下の通りです。

ヘッダー	説明
X-Rate-Limit-Remaining	アクセス可能な残り回数
X-Rate-Limit-Retry-After-Seconds	アクセス回数がリセットされるまでの時間（秒）※Rate Limitエラー時のみ

レスポンスJSON構造

成功時（200/201）

{
  "delivery_id": 1
}

エラー時（400/401/500など）

{
  "error_messages": {
    "main": ["エラーメッセージ"]
  }
}

レスポンスヘッダー取得例

def get_delivery_with_rate_limit_info(client: BlastEngineClient, delivery_id: int):
    """配信詳細とRate Limit情報を取得"""
    url = f"{client.base_url}/deliveries/{delivery_id}"
    headers = client._get_headers()

    response = client.session.get(url, headers=headers, timeout=client.timeout)
    response.raise_for_status()

    # Rate Limit情報の取得
    rate_limit_remaining = response.headers.get("X-Rate-Limit-Remaining")

    print(f"残りリクエスト数: {rate_limit_remaining}")

    return response.json()

HTTPステータスコード別の推奨対応

blastengine APIが返すHTTPステータスコードと、それぞれの推奨対応を整理します。

ステータスコード一覧

コード	種別	説明	推奨対応
200	OK	リクエスト成功	正常処理
201	Created	リソース作成成功	正常処理
400	Bad Request	リクエスト値が不正	パラメータを修正して再送
401	Unauthorized	認証エラー	Bearer Tokenを確認
403	Forbidden	API利用権限なし	アカウント権限を確認
404	Not Found	指定リソースが存在しない	リソースIDを確認
429	Too Many Requests	Rate Limit超過	X-Rate-Limit-Retry-After-Seconds秒待機後にリトライ
500	Internal Server Error	サーバエラー	リトライまたはサポートに問い合わせ
502	Bad Gateway	高負荷によるエラー	リトライ

Python実装例

import time
from typing import Optional, Dict, Any
import logging

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

class BlastEngineAPIError(Exception):
    """blastengine API エラー基底クラス"""
    pass

class RateLimitError(BlastEngineAPIError):
    """Rate Limitエラー"""
    def __init__(self, retry_after: int):
        self.retry_after = retry_after
        super().__init__(f"Rate limit exceeded. Retry after {retry_after} seconds.")

class BlastEngineClient:
    """blastengine API クライアント（エラーハンドリング対応）"""

    def __init__(self, bearer_token: str, timeout: int = 30):
        self.base_url = "https://app.engn.jp/api/v1"
        self.bearer_token = bearer_token
        self.timeout = timeout
        self.session = requests.Session()

    def _get_headers(self, content_type: str = "application/json") -> dict:
        return {
            "Authorization": f"Bearer {self.bearer_token}",
            "Content-Type": content_type,
            "Accept-Language": "ja-JP"
        }

    def _handle_response(self, response: requests.Response) -> dict:
        """レスポンスのハンドリング"""

        # 200番台: 成功
        if response.status_code in [200, 201]:
            logger.info(f"Request successful: {response.status_code}")
            return response.json()

        # 400: バリデーションエラー
        elif response.status_code == 400:
            error_data = response.json()
            error_msg = error_data.get("error_messages", {})
            logger.error(f"Validation error: {error_msg}")
            raise BlastEngineAPIError(f"Validation error: {error_msg}")

        # 401: 認証エラー
        elif response.status_code == 401:
            logger.error("Authentication failed. Check your Bearer Token.")
            raise BlastEngineAPIError("Authentication failed. Check your Bearer Token.")

        # 403: 権限エラー
        elif response.status_code == 403:
            logger.error("Permission denied. Check your API access rights.")
            raise BlastEngineAPIError("Permission denied.")

        # 404: リソースが存在しない
        elif response.status_code == 404:
            logger.error("Resource not found.")
            raise BlastEngineAPIError("Resource not found.")

        # 429: Rate Limit
        elif response.status_code == 429:
            retry_after = int(response.headers.get("X-Rate-Limit-Retry-After-Seconds", 60))
            logger.warning(f"Rate limit exceeded. Retry after {retry_after} seconds.")
            raise RateLimitError(retry_after)

        # 500/502: サーバーエラー
        elif response.status_code in [500, 502]:
            logger.error(f"Server error: {response.status_code}")
            raise BlastEngineAPIError(f"Server error: {response.status_code}")

        else:
            logger.error(f"Unexpected error: {response.status_code}")
            raise BlastEngineAPIError(f"Unexpected error: {response.status_code}")

    def request_with_retry(
        self,
        method: str,
        endpoint: str,
        data: Optional[Dict[Any, Any]] = None,
        params: Optional[Dict[str, Any]] = None,
        max_retries: int = 3
    ) -> dict:
        """リトライ機能付きリクエスト"""
        url = f"{self.base_url}/{endpoint}"
        headers = self._get_headers()

        for attempt in range(max_retries):
            try:
                if method.upper() == "GET":
                    response = self.session.get(url, headers=headers, params=params, timeout=self.timeout)
                elif method.upper() == "POST":
                    response = self.session.post(url, headers=headers, json=data, timeout=self.timeout)
                elif method.upper() == "PUT":
                    response = self.session.put(url, headers=headers, json=data, timeout=self.timeout)
                elif method.upper() == "DELETE":
                    response = self.session.delete(url, headers=headers, timeout=self.timeout)
                else:
                    raise ValueError(f"Unsupported HTTP method: {method}")

                return self._handle_response(response)

            except RateLimitError as e:
                if attempt < max_retries - 1:
                    logger.info(f"Waiting {e.retry_after} seconds before retry...")
                    time.sleep(e.retry_after)
                else:
                    raise

            except requests.exceptions.RequestException as e:
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt  # Exponential backoff: 1秒, 2秒, 4秒...
                    logger.warning(f"Request failed. Retrying in {wait_time} seconds... (Attempt {attempt + 1}/{max_retries})")
                    time.sleep(wait_time)
                else:
                    logger.error(f"Request failed after {max_retries} attempts.")
                    raise

# 使用例
client = BlastEngineClient(bearer_token=os.environ.get("BLASTENGINE_BEARER_TOKEN"))

try:
    # 配信一覧の取得
    result = client.request_with_retry("GET", "deliveries/all", params={"page": 1, "size": 10})
    print(f"Success: {result}")

except BlastEngineAPIError as e:
    print(f"API Error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

まとめ

blastengine APIを効果的に活用するためのポイント

1. 認証とセキュリティ

• Bearer Tokenは環境変数で管理し、コードに直接記述しない
• トークンの生成手順（SHA256→小文字化→Base64）を正確に実装

2. リクエスト設計

• 適切なタイムアウト設定（推奨：30秒）
• Rate Limit（500req/m）を考慮した実装
• リトライ戦略（Exponential Backoff）の実装

3. エラーハンドリング

• HTTPステータスコード別の適切な処理
• 429エラー時はX-Rate-Limit-Retry-After-Secondsを参照
• エラーレスポンスのJSON構造を考慮した処理

4. 本番環境での推奨事項

• セッションの再利用による効率化
• 接続プールの活用
• 適切なタイムアウトとリトライ設定
• エラー率やレスポンスタイムの監視

サンプルコード全体

完全な実装例は以下のようになります。

import os
import time
import logging
from typing import Optional, Dict, Any
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

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

class BlastEngineClient:
    """blastengine API クライアント（本番環境対応版）"""

    def __init__(
        self,
        bearer_token: str,
        timeout: int = 30,
        max_retries: int = 3
    ):
        self.base_url = "https://app.engn.jp/api/v1"
        self.bearer_token = bearer_token
        self.timeout = timeout

        # セッション設定
        self.session = requests.Session()

        # リトライ戦略
        retry_strategy = Retry(
            total=max_retries,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST", "PUT", "PATCH", "DELETE"]
        )

        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)

    def _get_headers(self, content_type: str = "application/json") -> dict:
        return {
            "Authorization": f"Bearer {self.bearer_token}",
            "Content-Type": content_type,
            "Accept-Language": "ja-JP"
        }

    def get_deliveries(self, page: int = 1, size: int = 100) -> dict:
        """配信一覧を取得"""
        endpoint = "deliveries/all"
        params = {"page": page, "size": size, "sort": "delivery_time:desc"}

        response = self.session.get(
            f"{self.base_url}/{endpoint}",
            headers=self._get_headers(),
            params=params,
            timeout=self.timeout
        )
        response.raise_for_status()
        return response.json()

    def send_transaction_email(self, email_data: dict) -> dict:
        """トランザクションメールを送信"""
        endpoint = "deliveries/transaction"

        response = self.session.post(
            f"{self.base_url}/{endpoint}",
            headers=self._get_headers(),
            json=email_data,
            timeout=self.timeout
        )
        response.raise_for_status()
        return response.json()

# 使用例
if __name__ == "__main__":
    # 環境変数からBearer Tokenを取得
    bearer_token = os.environ.get("BLASTENGINE_BEARER_TOKEN")

    if not bearer_token:
        raise ValueError("BLASTENGINE_BEARER_TOKEN environment variable is not set")

    client = BlastEngineClient(bearer_token)

    try:
        # 配信一覧の取得
        deliveries = client.get_deliveries(page=1, size=10)
        logger.info(f"Retrieved {len(deliveries.get('data', []))} deliveries")

    except requests.exceptions.HTTPError as e:
        logger.error(f"HTTP Error: {e}")
    except Exception as e:
        logger.error(f"Error: {e}")

この実装により、blastengine APIを安全かつ効率的に利用できます。