eBay OAuth-Pythonでアクセストークンを管理しよう

eBay OAuth 完全ガイド：Pythonでアクセストークン管理を堅牢に実装する

はじめに

本記事は、前回の入門編の続きとして、中級編のスタートを切ります。OAuth 2.0 認証から一連の出品機能APIまで踏み込み、実戦で通用するあなたの出品システムを構築して行きましょう。サンプルコードや実装のベストプラクティスなどを豊富に盛り込んでいます。

まずは、eBay APIの利用に必須となる OAuth 2.0 認証 について深掘りします。入門編ではAPI Explorerなどの既存ツールで直接トークンを取得しましたが、実稼働する堅牢なシステムを構築するには、単にトークンを取得するだけのスクリプトと、実際のアプリケーションの認証機構との間に、乗り越えるべき大きな隔たりがあります。

背景・なぜこれが重要か (Motivation)

「API を叩いたら 401 Unauthorized が返ってきた」。これはAPI 開発において最もよくあるエラーです。

eBay の User Access Token の有効期限は 2時間（120分）と比較的短く設定されています。

数万件の在庫をバッチ処理で更新している最中にトークンが失効した場合、処理が途中で落ちてしまうとデータの不整合が発生します。そのため、「リクエストの直前に有効期限を確認し、切れていれば（あるいは切れそうであれば）自動で Refresh Token を使って再取得する」という自己修復型の認証基盤を最初に構築しておくことが、システム全体の安定性に直結します。

基本的な使い方（ベースライン）：Refresh Token はどこから来る？

自動更新機構を作る前に、一番最初の疑問である 「そもそも refresh_token はどうやって手に入れるの？」 を解決しておきましょう。

eBay の User Token を取得するには、初回のみブラウザ経由でのユーザー同意（Authorization Code Grant）が必要です。手順は以下の通りです。

1. 同意URLの発行: 必要なスコープ（権限）を指定した eBay のログインURLを生成し、ブラウザで開きます。
2. ログインと許可: 出品を行う eBay アカウントでログインし、アプリへのアクセスを許可（Grant）します。
3. Authorization Code の取得: 設定した Redirect URI に遷移した際、URLのパラメータに付与される code の文字列をコピーします。
4. Refresh Token の取得（初回のみ）: その code を使って eBay のトークンエンドポイントを叩き、最初の access_token と refresh_token を取得します。

refresh_token の有効期限は通常18ヶ月と長いため、一度取得すればデータベースや環境変数に保存して使い回すことができます。本記事で実装するクラスは、「すでに取得済みの refresh_token を使って、2時間ごとに切れる access_token を自動で再取得し続ける」ためのものです。

実務で躓く場面・深いポイント (Core)

実務で躓くポイントは、「トークンの更新処理自体」よりも「いつ、どうやって更新するか」という状態管理です。

ここでは、Python の requests を拡張し、自動的にトークンの状態を管理する eBayTokenManager クラスを実装します。

堅牢な TokenManager の実装

# ebay_token_manager.py
import requests
import base64
import time
import threading
from typing import Optional

class eBayTokenManager:
    def __init__(self, client_id: str, client_secret: str, refresh_token: str, environment: str = "production"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.refresh_token = refresh_token
        
        # 環境の切り替え（次回連載のSandbox対応への布石）
        self.base_url = "https://api.ebay.com" if environment == "production" else "https://api.sandbox.ebay.com"
        
        self._access_token: Optional[str] = None
        self._expires_at: float = 0.0
        
        # スレッドセーフな更新のためのロック
        self._lock = threading.Lock()

    def _get_auth_header(self) -> str:
        """Client ID と Secret を Base64 エンコードして Basic 認証ヘッダを生成"""
        cred = f"{self.client_id}:{self.client_secret}".encode('utf-8')
        b64_cred = base64.b64encode(cred).decode('utf-8')
        return f"Basic {b64_cred}"

    def _refresh_access_token(self) -> None:
        """Refresh Token を用いて新しい Access Token を取得する"""
        url = f"{self.base_url}/identity/v1/oauth2/token"
        headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Authorization": self._get_auth_header()
        }
        data = {
            "grant_type": "refresh_token",
            "refresh_token": self.refresh_token
        }

        response = requests.post(url, headers=headers, data=data)
        response.raise_for_status() # 4xx/5xx エラー時に例外を送出
        
        token_data = response.json()
        self._access_token = token_data["access_token"]
        
        # 深いポイント: 有効期限のバッファとして、実際の期限より60秒前に失効判定する（通信遅延などのエッジケース対策）
        self._expires_at = time.time() + int(token_data["expires_in"]) - 60

    def get_token(self) -> str:
        """
        有効なアクセストークンを返す。
        期限が切れている場合は自動的にリフレッシュする。
        """
        # ロックを取得して、複数スレッドからの同時更新（Race Condition）を防ぐ
        with self._lock:
            # トークンが未取得、または期限切れ（バッファ含む）の場合
            if not self._access_token or time.time() >= self._expires_at:
                self._refresh_access_token()
                
        return self._access_token

なぜロック (threading.Lock) が必要なのか？

バッチ処理において、API の並列呼び出し（ThreadPoolExecutor など）を行う際、トークンが切れた瞬間に複数スレッドが同時に _refresh_access_token() を呼び出す可能性があります。

これにより、API Rate Limit（レート制限）に不必要に引っかかる、あるいは最新のトークンが上書きされ競合状態になるというバグが発生します。_lock を用いることで、最初の一つのスレッドだけが更新処理を行い、他のスレッドは安全に最新のトークンを利用できます。

使い方サンプル

作成したクラスは、以下のようにインスタンス化して使用します。API リクエストを送る直前に get_token() を呼ぶだけで、常に有効なトークンが保証されます。

# 使い方の例
if __name__ == "__main__":
    # 事前に取得した各種キーを設定（実務では環境変数から読み込むことを推奨）
    manager = eBayTokenManager(
        client_id="YOUR_CLIENT_ID",
        client_secret="YOUR_CLIENT_SECRET",
        refresh_token="YOUR_REFRESH_TOKEN",
        environment="production"
    )
    
    # トークンを取得（初回なので内部で自動的にリフレッシュ通信が走る）
    token = manager.get_token()
    print(f"取得したトークン: {token[:20]}...")  # セキュリティのため最初の20文字だけ表示
    
    # 2回目の呼び出し（有効期限内なので、通信は発生せずキャッシュされたトークンが即座に返る）
    token2 = manager.get_token()

    # 実際のAPIリクエストではこのようにAuthorizationヘッダ（Bearer）に渡す
    headers = {
        "Authorization": f"Bearer {token2}",
        "Content-Type": "application/json"
    }
    # response = requests.get("https://api.ebay.com/sell/inventory/v1/inventory_item", headers=headers)
    # print(response.json())

まとめ

本記事では、eBay API 開発の第一歩として、実務に耐えうる堅牢な OAuth 2.0 アクセストークン管理の実装方法を解説しました。

• ベースライン: grant_type="refresh_token" を用いた更新処理の自動化。
• 深いポイント: スレッドセーフな設計と、期限切れ直前のエッジケース（60秒バッファ）の考慮。

単に「API が叩けた」で満足せず、こうした基盤を最初に固めることで、今後の開発体験が劇的に向上します。

（今回はメモリ内での管理を実装しましたが、将来的に複数サーバーで分散処理を行う規模になった際は、Redis などを活用したトークンの一元管理も検討に値します。これについては連載の後半で扱う予定です。）

次のステップ

次回（#2）は、「eBay Sandbox環境の完全セットアップ」です。

本番環境を汚さずに API のテストを行うためのテストアカウント作成から、Python コード上で Production / Sandbox を環境変数 .env でシームレスに切り替える実装パターンを解説します。お楽しみに！