ゼロから実務まで学ぶAPI連携17ステップで、「APIって何？」から「連携の設計を任される人」まで、ひと続きに

# 使える道具があるか確認する
curl --version

# 誰でも使える練習用のAPIに、いま日本は何時かを聞いてみる
curl "https://worldtimeapi.org/api/timezone/Asia/Tokyo"

# -v で通信の中身を表示（> が送った内容、< が返ってきた内容）
curl -v "https://httpbin.org/get"

# ヘッダーだけを見たいとき（-I は HEAD リクエスト）
curl -I "https://httpbin.org/get"

# 取得（GET）。-X は省略してよい（curlの既定がGETのため）
curl "https://jsonplaceholder.typicode.com/posts/1"

# 一覧を条件つきで取得（クエリパラメータ）
curl "https://jsonplaceholder.typicode.com/posts?userId=1&_limit=3"

# 作成（POST）。-d でボディを送り、形式をヘッダーで伝える
curl -X POST "https://jsonplaceholder.typicode.com/posts" \
  -H "Content-Type: application/json" \
  -d '{"title":"はじめてのAPI","body":"テスト投稿","userId":1}'

# 削除（DELETE）
curl -X DELETE "https://jsonplaceholder.typicode.com/posts/1"

{
  "status": "ok",
  "total": 2,
  "users": [
    {
      "id": 12,
      "name": "山田太郎",
      "active": true,
      "department": { "code": "S01", "name": "営業部" },
      "tags": ["新規", "重点"]
    },
    {
      "id": 13,
      "name": "鈴木花子",
      "active": false,
      "department": { "code": "K02", "name": "経理部" },
      "tags": []
    }
  ]
}

# Python が入っていれば、追加インストール不要で整形できる
curl -s "https://jsonplaceholder.typicode.com/users/1" | python3 -m json.tool

# jq を使う場合（別途インストール。API業務では定番の道具）
curl -s "https://jsonplaceholder.typicode.com/users/1" | jq

# ほしい値だけ取り出す（jq）
curl -s "https://jsonplaceholder.typicode.com/users/1" | jq '.address.city'

# 一覧から、名前だけを並べる
curl -s "https://jsonplaceholder.typicode.com/users" | jq '.[].name'

# わざと404を返してもらう
curl -i "https://httpbin.org/status/404"

# ステータスコードだけを数字で取り出す（バッチ処理の判定でよく使う書き方）
curl -s -o /dev/null -w "%{http_code}\n" "https://httpbin.org/status/500"

# 自分が送ったヘッダーを、そのまま見せてもらう
curl -s "https://httpbin.org/headers" -H "X-My-Header: test123"

# 応答にかかった時間も測る
curl -s -o /dev/null -w "code=%{http_code} time=%{time_total}s\n" "https://httpbin.org/delay/2"

# 一覧（絞り込みつき）
curl -s "https://jsonplaceholder.typicode.com/posts?userId=1&_limit=2"

# 1件取得
curl -s "https://jsonplaceholder.typicode.com/posts/1"

# 一部更新（PATCH）
curl -s -X PATCH "https://jsonplaceholder.typicode.com/posts/1" \
  -H "Content-Type: application/json" \
  -d '{"title":"題名だけ差し替え"}'

# 親子の関係（1番の投稿に付いたコメント一覧）
curl -s "https://jsonplaceholder.typicode.com/posts/1/comments?_limit=2"

# ① APIキー（ヘッダーで送る。推奨）
curl -s "https://api.example.com/v1/users" \
  -H "X-API-Key: $API_KEY"

# ② Basic認証（curlが自動でBase64にしてくれる）
curl -s -u "myuser:mypassword" "https://api.example.com/v1/users"

# ③ Bearerトークン（いま最も多い形）
curl -s "https://api.example.com/v1/users" \
  -H "Authorization: Bearer $ACCESS_TOKEN"

# ④ OAuth 2.0 でトークンを取る（クライアントクレデンシャル方式＝機械同士の連携）
curl -s -X POST "https://api.example.com/oauth/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "grant_type=client_credentials" \
  -d "client_id=$CLIENT_ID" \
  -d "client_secret=$CLIENT_SECRET"

# 【雛形】これをコピーして、URLとヘッダーだけ差し替える
curl -s -i --max-time 30 \
  -X POST "https://api.example.com/v1/orders" \
  -H "Authorization: Bearer $ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json" \
  -d '{"product_id":"A-100","quantity":2}'

# 長いJSONは、ファイルに書いて渡すほうが確実（@はファイル指定の意味）
curl -s -X POST "https://api.example.com/v1/orders" \
  -H "Content-Type: application/json" \
  -d @order.json

# クエリに日本語や記号が入るときは、curlに変換させる（-G はGETで送る指定）
curl -s -G "https://api.example.com/v1/search" \
  --data-urlencode "q=東京 支店" \
  --data-urlencode "limit=10"

# 通信の詳細だけを見て、本文は捨てる
curl -v -o /dev/null "https://api.example.com/v1/ping"

# 応答が遅いのか、処理が遅いのかを測る
curl -s -o /dev/null -w "接続:%{time_connect}s 初回応答:%{time_starttransfer}s 合計:%{time_total}s\n" \
  "https://api.example.com/v1/orders"

# 証明書の問題を確認する（-k は検証を無視。原因調査のときだけ使う）
curl -v -k "https://api.example.com/v1/ping"

# プロキシ越しに叩く（社内ネットワークでよく必要になる）
curl -s -x "http://proxy.example.co.jp:8080" "https://api.example.com/v1/ping"

openapi: 3.0.3
info:
  title: 受注管理API          # このAPIの名前
  version: "1.0.0"
servers:
  - url: https://api.example.com/v1     # ← 基点URL（本番）
  - url: https://sandbox.example.com/v1 # ← 検証環境
paths:
  /orders:                    # ← エンドポイント
    get:                      # ← メソッド
      summary: 注文の一覧を取得する
      parameters:
        - name: status
          in: query           # ← クエリパラメータ（in: path ならパス）
          required: false
          schema: { type: string, enum: [pending, paid, shipped] }
      responses:
        "200":
          description: 成功
          content:
            application/json:
              schema:
                type: array
                items: { $ref: "#/components/schemas/Order" }
        "401": { description: 認証エラー }
components:
  schemas:
    Order:
      type: object
      required: [id, status]   # ← 必ず入っている項目
      properties:
        id:       { type: integer, example: 1001 }
        status:   { type: string,  example: "paid" }
        amount:   { type: number,  example: 12000 }
        note:     { type: string,  nullable: true }  # ← null が来る可能性あり

// async を付けた関数の中では await が使える
async function getUser(id) {
  const res = await fetch(`https://jsonplaceholder.typicode.com/users/${id}`);

  // ★重要：fetch は 404 や 500 でも例外を投げない。自分で確かめる
  if (!res.ok) {
    throw new Error(`APIエラー: ${res.status} ${res.statusText}`);
  }

  const data = await res.json();   // JSON文字列を、扱える形に変換する
  return data;
}

// 呼び出す側。失敗する前提で try/catch を必ず書く
async function main() {
  try {
    const user = await getUser(1);
    console.log(user.name, user.address.city);
  } catch (err) {
    console.error("取得に失敗しました:", err.message);
  }
}
main();

async function createPost(title, body) {
  const res = await fetch("https://jsonplaceholder.typicode.com/posts", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",        // 送る形式を伝える
      "Accept": "application/json",
      // "Authorization": `Bearer ${token}`      // 認証が要るAPIではこれを足す
    },
    body: JSON.stringify({ title, body, userId: 1 })  // ★オブジェクトを文字列に変換
  });

  if (!res.ok) {
    // エラーの本文にも、たいてい理由が書いてある。読まずに捨てない
    const text = await res.text();
    throw new Error(`${res.status}: ${text}`);
  }
  return await res.json();
}

async function fetchWithTimeout(url, options = {}, ms = 10000) {
  const ctrl = new AbortController();
  const timer = setTimeout(() => ctrl.abort(), ms);
  try {
    return await fetch(url, { ...options, signal: ctrl.signal });
  } catch (err) {
    if (err.name === "AbortError") {
      throw new Error(`${ms}ミリ秒以内に応答がありませんでした`);
    }
    throw err;
  } finally {
    clearTimeout(timer);   // 成功しても失敗しても、必ず後片付けする
  }
}

// 新しい環境なら、これだけでも書ける
// const res = await fetch(url, { signal: AbortSignal.timeout(10000) });

<form id="contactForm" novalidate>
  <label for="name">お名前</label>
  <input id="name" name="name" type="text" required maxlength="40">

  <label for="email">メールアドレス</label>
  <input id="email" name="email" type="email" required>

  <label for="qty">数量</label>
  <input id="qty" name="qty" type="number" min="1" max="99" value="1">

  <label for="message">お問い合わせ内容</label>
  <textarea id="message" name="message" rows="5" required></textarea>

  <button type="submit" id="submitBtn">送信する</button>
  <p id="formMsg" role="status" aria-live="polite"></p>
</form>

const form = document.getElementById("contactForm");
const btn  = document.getElementById("submitBtn");
const msg  = document.getElementById("formMsg");

form.addEventListener("submit", async (e) => {
  e.preventDefault();                       // ① 画面の再読み込みを止める

  const fd = new FormData(form);            // ② 値をまとめて取り出す
  const payload = {
    name:    String(fd.get("name")  || "").trim(),
    email:   String(fd.get("email") || "").trim(),
    qty:     Number(fd.get("qty")   || 1),
    message: String(fd.get("message") || "").trim()
  };

  const error = validate(payload);           // ③ 送る前に検証する
  if (error) { show(error, "ng"); return; }

  btn.disabled = true;                       // ④ 二重送信を防ぐ
  show("送信しています…", "");

  try {                                      // ⑤ 送る
    const res = await fetch("/api/contacts", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify(payload),
      signal: AbortSignal.timeout(15000)
    });

    if (res.status === 422 || res.status === 400) {
      const err = await res.json();          // 相手が返した項目別エラーを使う
      show(err.message || "入力内容をご確認ください。", "ng");
      return;
    }
    if (!res.ok) throw new Error(`status ${res.status}`);

    const data = await res.json();           // ⑥ 結果を出す
    show(`受け付けました（受付番号: ${data.id}）`, "ok");
    form.reset();

  } catch (err) {
    // 利用者には分かる言葉で、開発者にはログで詳細を
    console.error(err);
    show("送信できませんでした。時間をおいて、もう一度お試しください。", "ng");
  } finally {
    btn.disabled = false;                    // 成功でも失敗でも必ず戻す
  }
});

function validate(v) {
  if (!v.name)                       return "お名前を入力してください。";
  if (v.name.length > 40)            return "お名前は40文字以内で入力してください。";
  if (!/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(v.email))
                                     return "メールアドレスの形式をご確認ください。";
  if (!Number.isInteger(v.qty) || v.qty < 1 || v.qty > 99)
                                     return "数量は1〜99の整数で入力してください。";
  if (!v.message)                    return "お問い合わせ内容を入力してください。";
  return null;
}

function show(text, kind) {
  msg.textContent = text;
  msg.className = kind ? `is-${kind}` : "";
}

const fd = new FormData();
fd.append("title", "請求書");
fd.append("file", document.getElementById("file").files[0]);

const res = await fetch("/api/uploads", {
  method: "POST",
  body: fd            // ★ headers は書かない（ブラウザが自動で正しく付ける）
});

# 作業用のフォルダを作り、その中だけで使う環境を用意する（推奨）
python3 -m venv .venv
source .venv/bin/activate        # Windowsは .venv\Scripts\activate
pip install requests

import os
import requests

BASE = "https://jsonplaceholder.typicode.com"

def get_user(user_id: int) -> dict:
    url = f"{BASE}/users/{user_id}"
    # timeout は (接続, 読み取り) の秒数。★必ず付ける
    res = requests.get(url, timeout=(3.0, 10.0))
    res.raise_for_status()        # 4xx・5xx なら例外を投げてくれる
    return res.json()

if __name__ == "__main__":
    try:
        user = get_user(1)
        print(user["name"], user["address"]["city"])
    except requests.exceptions.Timeout:
        print("時間内に応答がありませんでした")
    except requests.exceptions.HTTPError as e:
        print(f"APIエラー: {e.response.status_code} {e.response.text[:200]}")
    except requests.exceptions.RequestException as e:
        print(f"通信エラー: {e}")

import os
import requests

class ApiClient:
    def __init__(self):
        # ★鍵はコードに書かず、環境変数から読む（STEP 15）
        token = os.environ.get("EXAMPLE_API_TOKEN")
        if not token:
            raise RuntimeError("環境変数 EXAMPLE_API_TOKEN が設定されていません")

        self.base = os.environ.get("EXAMPLE_API_BASE", "https://api.example.com/v1")
        self.s = requests.Session()
        self.s.headers.update({
            "Authorization": f"Bearer {token}",
            "Accept": "application/json",
            # 誰からの通信か分かるようにしておく（相手のサポートに好まれる）
            "User-Agent": "kodomo-batch/1.0 (system@example.co.jp)",
        })

    def get(self, path: str, **params) -> dict:
        res = self.s.get(f"{self.base}{path}", params=params, timeout=(3.0, 30.0))
        res.raise_for_status()
        return res.json()

    def post(self, path: str, payload: dict) -> dict:
        # json= を使うと Content-Type: application/json が自動で付く
        res = self.s.post(f"{self.base}{path}", json=payload, timeout=(3.0, 30.0))
        res.raise_for_status()
        return res.json()

if __name__ == "__main__":
    api = ApiClient()
    orders = api.get("/orders", status="paid", limit=20)   # → ?status=paid&limit=20
    print(f"{len(orders)}件 取得しました")

import csv
import logging
from datetime import datetime, timezone, timedelta

JST = timezone(timedelta(hours=9))     # ★時差の取り違えは定番の不具合

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler("import.log", encoding="utf-8"),
              logging.StreamHandler()],
)
log = logging.getLogger(__name__)

def main():
    api = ApiClient()
    today = datetime.now(JST).strftime("%Y-%m-%d")
    log.info("取り込み開始 date=%s", today)

    try:
        orders = api.get("/orders", date=today, limit=100)
    except Exception:
        log.exception("取得に失敗しました")   # 例外の詳細もログに残す
        raise                                 # 異常終了させ、監視で気づけるようにする

    # 相手の項目名を、自社の項目名に置き換えて保存する
    with open(f"orders_{today}.csv", "w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f)
        w.writerow(["注文番号", "金額", "状態", "備考"])
        for o in orders:
            w.writerow([
                o.get("id"),
                o.get("amount", 0),
                o.get("status", "unknown"),
                o.get("note") or "",        # null のとき空文字にする
            ])

    log.info("取り込み完了 件数=%d", len(orders))

if __name__ == "__main__":
    main()

import random
import time
import requests

RETRYABLE = {429, 500, 502, 503, 504}      # 再試行してよいコード

def request_with_retry(session, method, url, max_attempts=4, **kwargs):
    kwargs.setdefault("timeout", (3.0, 30.0))

    for attempt in range(1, max_attempts + 1):
        try:
            res = session.request(method, url, **kwargs)

            if res.status_code not in RETRYABLE:
                res.raise_for_status()      # 成功、または再試行しないエラー
                return res

            # 相手が「何秒待て」と言っているなら、それに従う（最優先）
            wait = float(res.headers.get("Retry-After", 0)) or _backoff(attempt)

        except (requests.exceptions.Timeout,
                requests.exceptions.ConnectionError):
            wait = _backoff(attempt)        # 届かなかった場合も再試行の対象

        if attempt == max_attempts:
            raise RuntimeError(f"{max_attempts}回試しましたが失敗しました: {url}")

        print(f"{attempt}回目が失敗。{wait:.1f}秒待って再試行します")
        time.sleep(wait)

def _backoff(attempt: int) -> float:
    base = 2 ** (attempt - 1)               # 1, 2, 4, 8 秒
    return base + random.uniform(0, 0.5)    # ばらつきを足して、殺到を防ぐ

import uuid

# ★鍵は「1つの注文につき1つ」。再試行のたびに作り直してはいけない
key = str(uuid.uuid4())

res = request_with_retry(
    session, "POST", f"{BASE}/orders",
    headers={"Idempotency-Key": key},
    json={"product_id": "A-100", "quantity": 2},
)

log.error(
    "API失敗 method=%s url=%s status=%s request_id=%s order_id=%s body=%s",
    "POST", url, res.status_code,
    res.headers.get("X-Request-Id", "-"),
    order_id,
    res.text[:200],          # 全部は出さない（個人情報と容量に配慮）
)

def fetch_all(api, path, **params):
    items, cursor, page = [], None, 0
    MAX_PAGES = 1000                       # ★暴走を止める安全装置

    while True:
        page += 1
        if page > MAX_PAGES:
            raise RuntimeError("ページ数が上限を超えました。条件を見直してください")

        q = dict(params, limit=100)
        if cursor:
            q["cursor"] = cursor

        data = api.get(path, **q)
        items.extend(data["items"])
        print(f"{page}ページ目 累計 {len(items)}件")

        cursor = data.get("next_cursor")
        if not cursor:                     # 次が無ければ終わり
            break
        time.sleep(0.2)                    # ★相手への思いやり（叩きすぎ防止）

    return items

res = session.get(url, timeout=(3, 30))

remaining = int(res.headers.get("X-RateLimit-Remaining", 999))
if remaining < 5:
    reset = float(res.headers.get("X-RateLimit-Reset", 60))
    print(f"残り{remaining}回。{reset:.0f}秒待って再開します")
    time.sleep(reset)

# 前回取った時刻をファイルやDBに保存しておき、次回はそれ以降だけを取る
last = load_last_run()                       # 例: "2026-08-13T02:00:00+09:00"
items = fetch_all(api, "/orders", updated_since=last)

# ★保存するのは「取得を始めた時刻」。終わった時刻にすると、
#   処理中に更新されたデータを取りこぼす
save_last_run(started_at)

import hashlib
import hmac
import os
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["WEBHOOK_SECRET"].encode()

@app.post("/hooks/payment")
def payment_hook():
    raw = request.get_data()                       # ★署名検証は「生のまま」で行う
    sig = request.headers.get("X-Signature", "")

    # ① 署名を検証する（本物の相手からか確かめる）
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig):     # ★==ではなくこれを使う
        abort(401)

    event = request.get_json()
    event_id = event.get("id")

    # ② 重複を弾く（同じ通知が何度も来る前提）
    if already_processed(event_id):
        return "", 200                             # 二重処理せず、成功として返す
    mark_processed(event_id)

    # ③ 重い処理はここでやらない。キューに積んで、すぐ返す
    enqueue(event)
    return "", 200

import responses     # pip install responses

@responses.activate
def test_get_order_success():
    responses.add(
        responses.GET, "https://api.example.com/v1/orders/1",
        json={"id": 1, "status": "paid", "amount": 1200}, status=200,
    )
    order = ApiClient().get("/orders/1")
    assert order["status"] == "paid"

@responses.activate
def test_get_order_server_error():
    # ★異常系こそテストする。500のとき正しく諦められるか
    responses.add(responses.GET, "https://api.example.com/v1/orders/1", status=500)
    with pytest.raises(Exception):
        ApiClient().get("/orders/1")

# Mac / Linux
export EXAMPLE_API_TOKEN="abc123..."

# Windows PowerShell
$env:EXAMPLE_API_TOKEN = "abc123..."

# .env ファイルを使う場合、必ず .gitignore に追記する
echo ".env" >> .gitignore

# 【確認】うっかり鍵をコミットしていないか、履歴を検索する
git log -p | grep -i -E "api[_-]?key|secret|token|password" | head

■ 事象
POST /v1/orders で 422 が返り、注文を登録できません。

■ 発生日時
2026-08-14 10:32:15 JST（複数回発生、直近5回すべて）

■ 環境
検証環境（sandbox）／ クライアント: Python requests 2.32

■ 送信したリクエスト（機密情報は伏せています）
POST https://sandbox.example.com/v1/orders
Headers: Authorization: Bearer eyJh****（末尾4桁 : a1b2）
Body: {"product_id":"A-100","quantity":2,"delivery_date":"2026-08-20"}

■ 返ってきたレスポンス
HTTP 422
X-Request-Id: 7f3a9c22-1b44-4e0d-9f2a-55e1c0d3a8b1
{"error":{"code":"INVALID_FIELD","field":"delivery_date"}}

■ 確認したいこと
delivery_date の受け付け可能な書式と、指定できる日付の範囲をご教示ください。
ドキュメント（v1.4、8ページ）には「YYYY-MM-DD」と記載されています。