renue

ARTICLE

クラウドサイン Web API統合の実装ガイド【2026年版】— 401/429自動リトライ×300秒トークンバッファ×ページネーション3重判定の本番アーキテクチャ

公開日: 2026/4/6

電子契約サービスのデファクト「クラウドサイン」にはWeb APIが用意されているが、公式ドキュメントはSwagger仕様ベースで本番品質の実装パターンまでは踏み込んでいない。本記事ではrenueが社内システムに組み込んでいるCloudSignClientの実装コードをもとに、アクセストークン自動更新・401リトライ・429レート制限対応・ページネーション・PDFダウンロードまで含むプロダクション実装を解説する。

クラウドサイン Web APIの基礎

クラウドサイン Web APIはRESTfulなHTTP APIで、管理画面から書類作成・取得・ダウンロード・合意締結証明書取得まで可能な操作をプログラム経由で実行できる。認証はクライアントID→アクセストークン交換方式で、トークン有効期間は1時間。

項目仕様
認証client_id → access_token (1時間有効)
認証ヘッダAuthorization: Bearer {token}
レート制限429エラー時は Retry-After ヘッダに従う
ページネーションpage / per_page (最大100)
書類取得GET /documents, /documents/{id}
PDFダウンロードGET /documents/{id}/files/{file_id}
合意締結証明書GET /documents/{id}/certificate

レイヤー1: クライアント初期化とSSL設定

プロダクション実装で最初に考えるべきは「ローカル開発環境と本番環境の切り替え」。ローカルでは自己署名証明書や社内プロキシでSSL検証を外したいケースが多いが、本番では必須にしたい。環境変数で切り替える実装が堅牢だ。

class CloudSignClient:
    def __init__(
        self,
        client_id: Optional[str] = None,
        base_url: Optional[str] = None,
    ):
        self.client_id = client_id or env_setting.CLOUDSIGN_CLIENT_ID
        self.base_url = base_url or env_setting.CLOUDSIGN_BASE_URL

        if not self.client_id:
            raise ValueError("CLOUDSIGN_CLIENT_ID must be set")

        # ローカル開発時のみSSL検証を無効化
        self.verify_ssl = env_setting.PY_ENV != "LOCAL"

        self._access_token: Optional[str] = None
        self._token_expires_at: float = 0

ポイントは_access_tokenと_token_expires_atをインスタンス変数として持ち、トークンをメモリキャッシュすること。リクエストごとにトークン再発行するとレート制限に抵触する。

レイヤー2: アクセストークンの自動更新(300秒バッファ)

トークン管理で最も多いバグが「有効期限ギリギリでリクエストを投げて401エラー」パターン。これを防ぐために失効300秒前に再取得するバッファ設計を採用する。

def _get_access_token(self) -> str:
    url = f"{self.base_url}/token"
    data = {"client_id": self.client_id}

    try:
        response = requests.post(url, data=data, verify=self.verify_ssl)
        response.raise_for_status()
        token_data = response.json()

        access_token = str(token_data.get("access_token", ""))
        if not access_token:
            raise CloudSignTokenError("No access_token in response")

        self._access_token = access_token
        expires_in = int(str(token_data.get("expires_in", 3600)))
        # 有効期限の300秒前から更新対象とする
        self._token_expires_at = time.time() + expires_in - 300

        return access_token

    except requests.exceptions.HTTPError as e:
        status_code = e.response.status_code if e.response else None
        raise CloudSignTokenError(f"Failed to obtain access token (status: {status_code})")

def _ensure_access_token(self) -> str:
    if self._access_token and time.time() < self._token_expires_at:
        return self._access_token
    return self._get_access_token()

300秒バッファの根拠

  • リクエスト処理時間: 大きなPDFダウンロードは10-30秒かかることがある
  • 時計ズレ: サーバー間の時刻差を考慮して数十秒の余裕が必要
  • 429リトライ: Retry-Afterで待機中にトークンが切れるケース
  • 300秒 = 5分: これらを合算してもまず足りるマージン

レイヤー3: 401/429自動リトライ

本番運用で最も頻発するエラーは「401 Unauthorized(トークン失効)」と「429 Too Many Requests(レート制限)」。両者を透過的にリトライする実装が_requestメソッドの真髄。

def _request(
    self,
    method: str,
    endpoint: str,
    params=None,
    json_data=None,
    return_raw: bool = False,
):
    url = f"{self.base_url}{endpoint}"

    def _send_request(use_fresh_token: bool = False):
        if use_fresh_token:
            # 401時はトークン期限切れの可能性があるため強制再取得
            self._access_token = None
            self._token_expires_at = 0
        headers = self._get_headers()
        return requests.request(
            method=method,
            url=url,
            headers=headers,
            params=params,
            json=json_data,
            verify=self.verify_ssl,
        )

    response = _send_request()

    # 429レート制限 → Retry-After秒待機してリトライ
    if response.status_code == 429:
        retry_after = int(response.headers.get("Retry-After", 60))
        api_logger.warning(f"Rate limited. Retrying after {retry_after} seconds.")
        time.sleep(retry_after)
        response = _send_request()

    # 401認証エラー → トークン強制再取得してリトライ
    if response.status_code == 401:
        api_logger.warning(f"401 received. Retrying with refreshed token: {endpoint}")
        response = _send_request(use_fresh_token=True)

    response.raise_for_status()

    if return_raw:
        return response.content
    return response.json()

この実装の勘所

  • クロージャで_send_requestを内部定義: 同じurl/method/paramsを使いまわせる
  • 401は強制再取得: キャッシュのトークンがサーバー側で無効化されている可能性
  • 429のRetry-After: デフォルト60秒にフォールバック
  • return_raw フラグ: PDFダウンロードではbytes、通常はJSON

レイヤー4: 全件取得のページネーション

CloudSign APIはper_page最大100の制限があるため、全書類を取得するにはページネーションが必要。ただし「終了判定」を誤ると無限ループに陥る。

def get_all_documents(
    self,
    status: Optional[str] = None,
    keyword: Optional[str] = None,
    created_from: Optional[str] = None,
    created_to: Optional[str] = None,
    max_pages: int = 100,
) -> list[dict]:
    all_documents: list[dict] = []
    page = 1

    while page <= max_pages:
        response = self.get_documents(
            page=page, per_page=100,
            status=status, keyword=keyword,
            created_from=created_from, created_to=created_to,
        )

        documents = response.get("documents", [])
        if not documents:
            break

        for doc in documents:
            if isinstance(doc, dict):
                all_documents.append(doc)

        # 取得件数がper_page未満 → 最終ページ
        if len(documents) < 100:
            break

        # total_countに達した → 最終ページ
        total_count = response.get("total_count")
        if isinstance(total_count, int) and total_count > 0 and len(all_documents) >= total_count:
            break

        page += 1

    return all_documents

3重の終了判定

  1. max_pages (デフォルト100)で上限ガード — 無限ループ防止
  2. 取得件数がper_page未満 → 最終ページ
  3. total_count到達 → 最終ページ(サーバーが total_count を返す場合のみ)

「どれか1つでも満たせば終了」にすることで、APIレスポンスの揺らぎや空配列パターンにも強くなる。

レイヤー5: 例外設計(Token / API 2層)

エラーハンドリングでは「トークン取得失敗」と「APIリクエスト失敗」を別例外にする設計が有効だ。呼び出し側はCloudSignTokenErrorを「設定ミス/サーバー側障害」として扱い、CloudSignAPIErrorを「書類ID不正/権限不足/一時的障害」として扱える。

class CloudSignTokenError(Exception):
    # クラウドサイントークン取得エラー
    pass

class CloudSignAPIError(Exception):
    def __init__(
        self,
        message: str,
        status_code: Optional[int] = None,
        response_body: Optional[dict] = None,
    ):
        super().__init__(message)
        self.status_code = status_code
        self.response_body = response_body

CloudSignAPIErrorにはstatus_codeとresponse_bodyを持たせることで、呼び出し側で「404なら新規作成にフォールバック」「403ならSlack通知」のような分岐が書きやすくなる。

運用Tips: 合意締結証明書の永続化

実務で最も重要なのが合意締結証明書(certificate)のダウンロードと永続化だ。クラウドサイン側でも保管されているが、社内の法務/経理ワークフローの独立性を保つために自社ストレージ(Azure Blob/GCS/S3)にも冷凍保存するのが鉄則。

def archive_document(client: CloudSignClient, document_id: str, file_id: str):
    # 書類PDF本体
    pdf_bytes = client.download_file(document_id, file_id)
    blob_client.upload(f"contracts/{document_id}/main.pdf", pdf_bytes)

    # 合意締結証明書(タイムスタンプと署名情報入り)
    cert_bytes = client.get_certificate(document_id)
    blob_client.upload(f"contracts/{document_id}/certificate.pdf", cert_bytes)

まとめ

  1. トークンは300秒バッファで自動更新(リクエスト処理時間と時計ズレを吸収)
  2. 401自動リトライでサーバー側トークン失効に対応
  3. 429自動リトライでRetry-Afterに従う
  4. ページネーションは3重の終了判定で無限ループ防止
  5. 例外は Token / API の2層で呼び出し側の分岐を簡潔に
  6. 合意締結証明書は自社ストレージにも冷凍保存

この設計でrenueは契約書の一括インポート・AI分析・社内検索システムを本番運用している。電子契約APIは「つながる」だけでは不十分で、長期運用に耐える堅牢性が鍵だ。

よくある質問

Q. リトライ回数に上限は?

この実装では401/429それぞれ最大1回のリトライ。指数バックオフで複数回リトライする場合はtenacityライブラリを組み合わせるとよい。

Q. 並行リクエストは?

クラウドサインのレート制限は比較的緩いが、並行実行時はセマフォで同時実行数を5-10程度に制限することを推奨。

Q. Webhookには対応している?

クラウドサインは書類ステータス変更のWebhookを提供している。ポーリングよりWebhook優先で実装すると無駄なAPIコールが減る。

navigate_before記事一覧へ戻る