API入門：OpenAPI/HTTPの基本と“壊れない”Pythonクライアント設計（コピペOK）

from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class ApiError(Exception): ...
class RateLimitError(ApiError): ...
class ExternalError(ApiError): ...
class DataError(ApiError): ...

UA = "pythonbunseki-api/1.0 (+https://pythonbunseki.com)"

@dataclass(slots=True)
class ApiClient:
    base_url: str
    api_key: str | None = None
    oauth_token: str | None = None
    timeout: float = 10.0

    def __post_init__(self):
        s = requests.Session()
        s.headers.update({
            "User-Agent": UA,
            "Accept": "application/json",
            "Accept-Language": "ja,en;q=0.8",
        })
        if self.api_key:
            s.headers["X-API-Key"] = self.api_key
        if self.oauth_token:
            s.headers["Authorization"] = f"Bearer {self.oauth_token}"
        retry = Retry(
            total=3,
            backoff_factor=0.7,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "HEAD"],
            respect_retry_after_header=True,
        )
        s.mount("https://", HTTPAdapter(max_retries=retry))
        s.mount("http://", HTTPAdapter(max_retries=retry))
        self._s = s

    def get(self, path: str, params: dict[str, Any] | None = None, *, etag: str | None = None) -> tuple[dict, str | None]:
        url = self.base_url.rstrip("/") + "/" + path.lstrip("/")
        headers = {}
        if etag:
            headers["If-None-Match"] = etag
        r = self._s.get(url, params=params, headers=headers, timeout=self.timeout)
        if r.status_code == 304:
            raise ExternalError("not modified")
        if r.status_code == 429:
            raise RateLimitError(r.text)
        try:
            r.raise_for_status()
        except requests.HTTPError as e:
            raise ExternalError(f"{r.status_code}: {r.text[:200]}") from e
        etag_new = r.headers.get("ETag")
        try:
            data = r.json()
        except ValueError as e:
            raise DataError("invalid json") from e
        return data, etag_new

from typing import Any, Iterator

# A) page/limit 方式

def iter_pages_page_limit(cli: ApiClient, path: str, *, limit: int = 100) -> Iterator[dict]:
    page = 1
    while True:
        data, _ = cli.get(path, {"page": page, "limit": limit})
        items: list[dict[str, Any]] = data.get("items", [])
        if not items:
            break
        for it in items:
            yield it
        page += 1

# B) cursor/next_token 方式

from typing import Iterator

def iter_pages_cursor(cli: ApiClient, path: str, *, cursor: str | None = None) -> Iterator[dict]:
    token = cursor
    while True:
        params = {"cursor": token} if token else None
        data, _ = cli.get(path, params)
        for it in data.get("data", []):
            yield it
        token = data.get("next")
        if not token:
            break

# C) Linkヘッダ方式（<...>; rel="next"）

def parse_link_next(link_header: str | None) -> str | None:
    if not link_header:
        return None
    for part in link_header.split(","):
        seg = part.strip().split(";")
        if len(seg) >= 2 and 'rel="next"' in seg[1]:
            url = seg[0].strip()
            if url.startswith("<") and url.endswith(">"):
                return url[1:-1]
    return None

# 実運用ではレスポンスヘッダから次URLをたどる実装にする

# ETagキャッシュ（例）
etag_cache: dict[str, str] = {}

def fetch_users(cli: ApiClient) -> list[dict]:
    etag = etag_cache.get("/users")
    try:
        data, etag_new = cli.get("/users", etag=etag)
    except ExternalError as e:
        if str(e) == "not modified":
            return []  # 差分なし
        raise
    if etag_new:
        etag_cache["/users"] = etag_new
    return data.get("items", [])

import hmac, hashlib, base64, time

def signed_headers(secret: str, method: str, path: str, body: bytes = b"") -> dict[str, str]:
    ts = str(int(time.time()))
    msg = (method.upper() + path + ts).encode() + body
    sig = hmac.new(secret.encode(), msg, hashlib.sha256).digest()
    return {"X-Timestamp": ts, "X-Signature": base64.b64encode(sig).decode()}

from pathlib import Path
import json, pandas as pd

def save_ndjson(items: list[dict], path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as f:
        for it in items:
            f.write(json.dumps(it, ensure_ascii=False) + "\n")

def save_parquet(items: list[dict], path: Path) -> None:
    import pandas as pd
    df = pd.DataFrame(items)
    df.to_parquet(path, compression="zstd")

openapi: 3.0.3
info: { title: Sample API, version: '1.0' }
servers: [ { url: https://api.example.com } ]
paths:
  /users:
    get:
      summary: List users
      parameters:
        - in: query
          name: page
          schema: { type: integer, minimum: 1 }
        - in: query
          name: limit
          schema: { type: integer, maximum: 100 }
      responses:
        '200':
          description: OK
          content:
            application/json:
              schema:
                type: object
                properties:
                  items:
                    type: array
                    items:
                      $ref: '#/components/schemas/User'
  /users/{id}:
    get:
      parameters:
        - in: path
          name: id
          required: true
          schema: { type: string }
      responses:
        '200': { $ref: '#/components/responses/User' }
components:
  schemas:
    User:
      type: object
      required: [id, name]
      properties:
        id: { type: string }
        name: { type: string }
        email: { type: string, nullable: true }

import uuid

def post_json(cli: ApiClient, path: str, payload: dict) -> dict:
    url = cli.base_url.rstrip("/") + "/" + path.lstrip("/")
    headers = {"Idempotency-Key": uuid.uuid4().hex}
    r = cli._s.post(url, json=payload, headers=headers, timeout=cli.timeout)
    r.raise_for_status()
    return r.json()

from client import ApiClient

class DummyResp:
    def __init__(self, status_code=200, json_data=None, headers=None):
        self.status_code = status_code
        self._json = json_data or {}
        self.headers = headers or {}
        self.text = ""
        self.encoding = "utf-8"
    def json(self):
        return self._json
    def raise_for_status(self):
        if not (200 <= self.status_code < 300):
            raise Exception("http error")

class DummySession:
    def __init__(self, resps):
        self._resps = resps
        self.headers = {}
    def get(self, *a, **k):
        return self._resps.pop(0)

def test_pagination_page_limit(monkeypatch):
    cli = ApiClient("https://api.example.com")
    resps = [
        DummyResp(json_data={"items":[{"id":1},{"id":2}]}),
        DummyResp(json_data={"items":[]}),
    ]
    cli._s = DummySession(resps)
    from client import iter_pages_page_limit
    ids = [it["id"] for it in iter_pages_page_limit(cli, "/users")]
    assert ids == [1,2]