コンテンツにスキップ

REST 演習3:Python でページングしながら全件取得する

  • hasMore を見て、全ページを取り切るループを書ける
  • 取得結果を、Excel で開ける CSV(UTF-8 BOM 付き)に書き出せる
  • 429・5xx は再試行し、4xx は再試行しない、という判断をコードにできる(Retry-After、指数バックオフ)
  • タイムアウトと資格情報(環境変数)を、正しく扱える
  • 実機に持っていくときに、何を変えるかを説明できる

検索と並べ替えは REST 演習1、作成・更新とエラーは REST 演習2 で扱いました。短いコード例は REST API の基本 と REST API 応用 にもあります。この演習は、それらを 1 本の実用的なスクリプトにまとめます。

前提・準備(Fusion 環境あり/なし)

Section titled “前提・準備(Fusion 環境あり/なし)”

サンプル商事株式会社の経理部が、毎晩のバッチで請求書の一覧を CSV に出し、Excel や BI ツールに取り込みます。途中で通信が不安定になっても、夜間に止まらず、前日の CSV を壊さないスクリプトが必要です。

環境なしルート(この演習の標準)

Section titled “環境なしルート(この演習の標準)”

練習用モック API と Python を使います。Fusion 環境は要りません。

  • Python 3.9 以降と、requests ライブラリを使います
  • curl と jq は、この演習では必須ではありません。CSV の確認には head と wc を使います
  • Python を実行できない場合は、練習用 API プレイグラウンド の入力例「障害を起こす:429(Retry-After 付き)」を送り、応答のステータスと Retry-After を確認してください。コードを読んで、解答例の出力と照らし合わせるだけでも学べます
Python の準備(初回のみ)
python3 -m venv .venv
source .venv/bin/activate # Windows の Git Bash では source .venv/Scripts/activate
pip install requests

接続先と資格情報は、環境変数で渡します(コードには書きません)。モックは、どんなユーザー名とパスワードでも通します。

ターミナルウィンドウ
export API_BASE="https://<このサイトのURL>/api/mock/fscmRestApi/resources/11.13.18.05"
export API_USER=learner
export API_PASSWORD=welcome1

<このサイトのURL> は、このサイトのアドレスに置き換えます。

同じスクリプトを、実機の請求書(invoices)に向けられます。変えるのは環境変数だけです。

  • API_BASE を https://<your-pod>.fa.<region>.oraclecloud.com/fscmRestApi/resources/11.13.18.05 にします
  • API_USER と API_PASSWORD を、テスト環境の統合ユーザーにします。パスワードは、read -rs API_PASSWORD; export API_PASSWORD のように、画面にも履歴にも残らない方法で設定します
  • 読み取り(GET)だけなので、データは変わりません。ただし、大量の取得は環境に負荷をかけます。必ずテスト環境で、PAGE_SIZE を小さくして試します

実機で変える点の全体は、最後の「実機との違い」の表にまとめています。

ステップ1 最小のループで全件を取る

Section titled “ステップ1 最小のループで全件を取る”

まず、動く最小のコードです。min_fetch.py として保存します。

min_fetch.py
import os
import requests
BASE = os.environ["API_BASE"]
AUTH = (os.environ["API_USER"], os.environ["API_PASSWORD"])
items, offset = [], 0
while True:
r = requests.get(
f"{BASE}/invoices",
auth=AUTH,
timeout=30,
params={"limit": 10, "offset": offset, "orderBy": "InvoiceId",
"onlyData": "true", "fields": "InvoiceId,InvoiceNumber,InvoiceAmount"},
)
r.raise_for_status()
page = r.json()
items += page["items"]
print(f"offset={offset} count={page['count']} hasMore={page['hasMore']}")
if not page["hasMore"]:
break
offset += page["count"]
print(len(items), "件")
ターミナルウィンドウ
python3 min_fetch.py
実行結果
offset=0 count=10 hasMore=True
offset=10 count=10 hasMore=True
offset=20 count=10 hasMore=False
30 件

押さえる点は 4 つです。

  • hasMore が False になるまで繰り返します。「count が limit より少ない」ではなく、hasMore を終了の条件にします
  • offset は、limit ではなく、返ってきた count ずつ進めます。サーバーが要求より少ない件数を返しても、取りこぼしません
  • orderBy=InvoiceId で並びを固定します。固定しないと、ページの間で重複や抜けが起きることがあります
  • timeout は必ず付けます。requests は、timeout を指定しないと待ち続けます(requests の公式ドキュメントも、「timeout を明示しなければタイムアウトしない」と説明しています)

このコードは動きますが、実務には足りません。一時的なエラーで、途中で落ちます。エラー時に、応答の本文(原因のメッセージ)が見えません。結果がメモリ上にしかありません。環境変数が未設定のときは、KeyError で止まります。

再試行、CSV 出力、タイムアウト、資格情報の扱いを足します。fetch_invoices.py として保存します。

fetch_invoices.py
import csv
import logging
import os
import random
import sys
import time
import requests
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("fetch")
def env(name: str) -> str:
"""必須の環境変数を読む。なければ、分かりやすいメッセージで終了する。"""
value = os.environ.get(name)
if not value:
sys.exit(f"環境変数 {name} を設定してください")
return value
BASE = env("API_BASE").rstrip("/")
AUTH = (env("API_USER"), env("API_PASSWORD")) # 資格情報はコードに書かず、環境変数から読む
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "10"))
MAX_ATTEMPTS = int(os.environ.get("MAX_ATTEMPTS", "5"))
TIMEOUT = (5, float(os.environ.get("READ_TIMEOUT", "30"))) # (接続, 読み取り) の秒数。必ず指定する
RETRYABLE = {408, 429, 500, 502, 503, 504} # 一時的なエラー。400・401・403・404 などは再試行しない
HEADERS = {"REST-Framework-Version": "2"} # Version は明示して固定する
# 練習用:モックに障害や遅延を起こさせるヘッダー(実機では不要)
for name, header in (("MOCK_FAULT", "X-Mock-Fault"), ("MOCK_DELAY", "X-Mock-Delay")):
if os.environ.get(name):
HEADERS[header] = os.environ[name]
FIELDS = ["InvoiceId", "InvoiceNumber", "Supplier", "InvoiceDate", "InvoiceCurrency", "InvoiceAmount", "PaidStatus"]
def retry_after(response: requests.Response):
"""Retry-After が秒数ならその値、なければ None。"""
value = response.headers.get("Retry-After", "")
return min(float(value), 60) if value.replace(".", "", 1).isdigit() else None
def get_json(session: requests.Session, url: str, params: dict) -> dict:
"""GET して JSON を返す。一時的なエラーは再試行し、それ以外のエラーはすぐに失敗させる。"""
for attempt in range(1, MAX_ATTEMPTS + 1):
try:
r = session.get(url, params=params, timeout=TIMEOUT)
except (requests.ConnectionError, requests.Timeout) as e:
reason, delay = type(e).__name__, None
else:
if r.status_code < 400:
return r.json()
if r.status_code not in RETRYABLE:
raise RuntimeError(f"HTTP {r.status_code}: {r.text[:300]}") # 再試行しても直らない
reason, delay = f"HTTP {r.status_code}", retry_after(r)
if attempt == MAX_ATTEMPTS:
raise RuntimeError(f"{url} が {MAX_ATTEMPTS} 回失敗しました(最後の原因: {reason})")
if delay is None:
delay = min(2 ** attempt, 30) # 指数バックオフ:2, 4, 8, 16 秒…
log.warning(" %s: %d 秒後に再試行します(%d/%d)", reason, delay, attempt, MAX_ATTEMPTS)
time.sleep(delay + random.random()) # ゆらぎ(0〜1 秒)を足して、再試行が重ならないようにする
def fetch_all(session: requests.Session, resource: str, params: dict):
"""hasMore が false になるまで offset を進めながら、全ページの行を 1 件ずつ返す。"""
offset = 0
while True:
page = get_json(session, f"{BASE}/{resource}", {**params, "limit": PAGE_SIZE, "offset": offset})
log.info("offset=%d: %d 件(hasMore=%s)", offset, page["count"], page["hasMore"])
yield from page["items"]
if not page["hasMore"] or page["count"] == 0: # count == 0 の確認は、無限ループの保険
return
offset += page["count"]
def main() -> None:
out = sys.argv[1] if len(sys.argv) > 1 else "invoices.csv"
params = {"orderBy": "InvoiceId", "onlyData": "true", "fields": ",".join(FIELDS)}
if len(sys.argv) > 2:
params["q"] = sys.argv[2] # 例:"PaidStatus = 'Unpaid'"
total = 0
part = out + ".part" # 途中で失敗したときに、前回の CSV を壊さないための一時ファイル
with requests.Session() as session, open(part, "w", newline="", encoding="utf-8-sig") as f:
session.auth = AUTH
session.headers.update(HEADERS)
writer = csv.DictWriter(f, fieldnames=FIELDS, extrasaction="ignore")
writer.writeheader()
for row in fetch_all(session, "invoices", params):
writer.writerow(row)
total += 1
os.replace(part, out) # 最後まで取れたときだけ、本来のファイル名にする
log.info("%d 件を %s に書きました", total, out)
if __name__ == "__main__":
try:
main()
except (RuntimeError, requests.RequestException) as e:
sys.exit(f"失敗: {e}")

get_json() の判断は、次の表のとおりです。

結果 動作
ステータスが 400 未満 JSON を返す
408・429・500・502・503・504、接続エラー、タイムアウト 再試行する。Retry-After があればその秒数、なければ 2・4・8・16 秒(指数バックオフ)に、ゆらぎを足して待つ
それ以外の 4xx・5xx(400・401・403・404 など) 再試行せず、応答の本文つきですぐ失敗させる
  • 公式の Status Codes には、再試行の対象にしたコードのうち、408・500・503・504 が載っています。429 と 502 は載っていません。429 は、このモックが Retry-After 付きで返すコードです。一般的な REST API では呼び出し回数の制限で返るので、受け取ったら Retry-After に従う実装にしておきます
  • fetch_all() は、行を 1 件ずつ返すジェネレーターです。全件をメモリに溜めずに、CSV へ書けます
  • CSV は utf-8-sig(BOM 付き)で書くので、日本語環境の Excel でも文字化けしません。newline="" は、csv モジュールを使うときの決まりです
  • CSV は、まず .part に書き、最後まで取れたときだけ os.replace() で本来の名前にします

ステップ3 実行して CSV を確認する

Section titled “ステップ3 実行して CSV を確認する”
ターミナルウィンドウ
python3 fetch_invoices.py
head -n 4 invoices.csv
wc -l < invoices.csv
実行結果
offset=0: 10 件(hasMore=True)
offset=10: 10 件(hasMore=True)
offset=20: 10 件(hasMore=False)
30 件を invoices.csv に書きました
InvoiceId,InvoiceNumber,Supplier,InvoiceDate,InvoiceCurrency,InvoiceAmount,PaidStatus
300100001,SMP-0001,サンプル商事株式会社,2026-01-10,JPY,485000,Paid
300100002,1001,仮想オフィス用品株式会社,2026-01-15,JPY,64800,Paid
300100003,1001,テスト物産株式会社,2026-01-20,JPY,760000,Paid
31

31 行は、見出しの 1 行と、請求書 30 件です。

ステップ4 一時的なエラーで再試行を確認する

Section titled “ステップ4 一時的なエラーで再試行を確認する”

モックのヘッダー X-Mock-Fault で、障害を起こします。random は、約 3 回に 1 回、429 か 503 を返します(どちらも Retry-After: 1 付きです)。

ターミナルウィンドウ
MOCK_FAULT=random python3 fetch_invoices.py random.csv
出力例(実行ごとに異なります)
offset=0: 10 件(hasMore=True)
HTTP 503: 1 秒後に再試行します(1/5)
HTTP 503: 1 秒後に再試行します(2/5)
offset=10: 10 件(hasMore=True)
HTTP 429: 1 秒後に再試行します(1/5)
HTTP 503: 1 秒後に再試行します(2/5)
offset=20: 10 件(hasMore=False)
30 件を random.csv に書きました

再試行が入っても、30 件がそろいます。次に、必ず失敗する 429 を送ります。5 回失敗して、あきらめます(約 7 秒かかります)。このとき、前回の invoices.csv が壊れていないことも確認します。

ターミナルウィンドウ
MOCK_FAULT=429 python3 fetch_invoices.py invoices.csv
ls -1 invoices.csv*
wc -l < invoices.csv
実行結果
HTTP 429: 1 秒後に再試行します(1/5)
HTTP 429: 1 秒後に再試行します(2/5)
HTTP 429: 1 秒後に再試行します(3/5)
HTTP 429: 1 秒後に再試行します(4/5)
失敗: https://<このサイトのURL>/api/mock/fscmRestApi/resources/11.13.18.05/invoices が 5 回失敗しました(最後の原因: HTTP 429)
invoices.csv
invoices.csv.part
31

invoices.csv は 31 行のままで、途中までの内容は invoices.csv.part に残っています。最後に、再試行してはいけない 400 を起こします。Version 2 では文字列を引用符で囲む必要があるので、引用符なしの q を渡します。

ターミナルウィンドウ
python3 fetch_invoices.py bad.csv "PaidStatus = Unpaid"
実行結果
失敗: HTTP 400: Unpaid is not a valid value. Enclose string values in single quotes, for example PaidStatus = 'Unpaid'.

再試行のログは出ずに、すぐ失敗します。同じ要求は同じ理由で失敗するので、待っても無駄です。応答の本文(原因のメッセージ)を、そのまま表示しています。

ステップ5 タイムアウトと資格情報を確認する

Section titled “ステップ5 タイムアウトと資格情報を確認する”

X-Mock-Delay で、モックの応答を 3 秒遅らせます。読み取りのタイムアウトを 1 秒にすると、ReadTimeout になります(再試行も 2 回で打ち切ります)。

ターミナルウィンドウ
MOCK_DELAY=3000 READ_TIMEOUT=1 MAX_ATTEMPTS=2 python3 fetch_invoices.py slow.csv
実行結果
ReadTimeout: 2 秒後に再試行します(1/2)
失敗: https://<このサイトのURL>/api/mock/fscmRestApi/resources/11.13.18.05/invoices が 2 回失敗しました(最後の原因: ReadTimeout)

タイムアウトを既定の 30 秒に戻すと、3 ページ分で約 9 秒かかりますが、成功します。

ターミナルウィンドウ
MOCK_DELAY=3000 python3 fetch_invoices.py slow.csv
実行結果
offset=0: 10 件(hasMore=True)
offset=10: 10 件(hasMore=True)
offset=20: 10 件(hasMore=False)
30 件を slow.csv に書きました

接続のタイムアウト(5 秒)と、読み取りのタイムアウト(30 秒)は別の値です。実機は、条件によって応答に時間がかかります。ReadTimeout や 504 が出るときは、PAGE_SIZE を小さくし、fields で項目を絞り、READ_TIMEOUT を調整します。

最後に、環境変数が足りないときの動作を確認します。

ターミナルウィンドウ
env -u API_PASSWORD python3 fetch_invoices.py
実行結果
環境変数 API_PASSWORD を設定してください

資格情報を、コードにも、コマンドラインの引数にも書かないのが原則です。引数は、プロセス一覧や実行履歴に残ります。ログにも、パスワードやトークンを出さないでください。

  • 最小版で、30 件を 3 ページ(offset が 0・10・20)で取得できた
  • offset を count ずつ進める理由と、hasMore で止める理由を説明できる
  • CSV が 31 行(見出し + 30 件)で、Excel で文字化けせずに開ける
  • MOCK_FAULT=random で、再試行のログが出ても 30 件がそろった
  • MOCK_FAULT=429 で失敗したあとも、前回の invoices.csv が壊れていない
  • 400 のときは、再試行せずに、原因のメッセージを表示して失敗した
  • READ_TIMEOUT=1 で ReadTimeout になり、既定値に戻すと成功した
  • 環境変数が足りないとき、分かりやすいメッセージで止まった

PAGE_SIZE=7 で実行し、何回リクエストするか、各回の件数はいくつかを答えてください。

解答例を表示
ターミナルウィンドウ
PAGE_SIZE=7 python3 fetch_invoices.py page7.csv
実行結果
offset=0: 7 件(hasMore=True)
offset=7: 7 件(hasMore=True)
offset=14: 7 件(hasMore=True)
offset=21: 7 件(hasMore=True)
offset=28: 2 件(hasMore=False)
30 件を page7.csv に書きました

5 回です。最後のページは 2 件で、hasMore が False なので、そこで止まります。offset を count ずつ進めているので、最後の offset は 28 です。

課題2 未払いの請求書だけを CSV にする

Section titled “課題2 未払いの請求書だけを CSV にする”

支払状況が Unpaid の請求書だけを、unpaid.csv に出力してください。スクリプトは書き換えず、第 2 引数に q を渡します。

解答例を表示

Version 2 の q なので、文字列を単一引用符で囲みます。

ターミナルウィンドウ
python3 fetch_invoices.py unpaid.csv "PaidStatus = 'Unpaid'"
wc -l < unpaid.csv
実行結果
offset=0: 10 件(hasMore=True)
offset=10: 4 件(hasMore=False)
14 件を unpaid.csv に書きました
15

14 件(見出しを含めて 15 行)です。絞り込みは、サーバー側の q で行います。全件を取ってから Python で絞ると、無駄な転送が増えます。

課題3 Retry-After がない 500 の待ち時間

Section titled “課題3 Retry-After がない 500 の待ち時間”

MOCK_FAULT=500 は、Retry-After を付けずに 500 を返します。MAX_ATTEMPTS=4 で実行し、待ち時間がどう増えるかを記録してください。

解答例を表示
ターミナルウィンドウ
MOCK_FAULT=500 MAX_ATTEMPTS=4 python3 fetch_invoices.py fail500.csv
実行結果
HTTP 500: 2 秒後に再試行します(1/4)
HTTP 500: 4 秒後に再試行します(2/4)
HTTP 500: 8 秒後に再試行します(3/4)
失敗: https://<このサイトのURL>/api/mock/fscmRestApi/resources/11.13.18.05/invoices が 4 回失敗しました(最後の原因: HTTP 500)

待ち時間は、2 秒、4 秒、8 秒と倍に増えます(これが指数バックオフです)。実際には、各回にゆらぎ(0〜1 秒)が足されるので、合計は約 14〜17 秒です。Retry-After がある 429 や 503 では、その秒数(1 秒)に従います。

症状 原因と対処
環境変数 … を設定してください export を忘れた、または別のターミナルで実行している。同じシェルで export してから実行する
ModuleNotFoundError: No module named 'requests' requests が未導入。仮想環境を有効にして pip install requests を実行する
ConnectionError の再試行が続く API_BASE の誤り(ホスト名・パス)、またはネットワーク・プロキシの設定。URL を curl で確かめる
失敗: HTTP 400: … 要求の誤り。メッセージを読んで、q・fields・orderBy を直す。再試行しない
失敗: HTTP 401: … 資格情報の誤り。環境変数の値を確認する(実機では、期限切れやアカウントのロックもある)
429・503 が続いて失敗する Retry-After に従っても続くなら、PAGE_SIZE を下げる、実行の間隔を空ける。MAX_ATTEMPTS を増やすだけでは直らない
CSV の日本語が Excel で文字化けする encoding="utf-8-sig" で書いているか確認する。BOM が邪魔な環境では utf-8 に変える
件数が合わない・重複する orderBy がない、または取得中にデータが更新された。orderBy に一意の属性を付ける。更新が多いなら、キーセット方式を使う(発展課題1)
  1. キーセット方式にする:offset を使わず、「前のページの最後の InvoiceId より大きいもの」を q で指定して取ります。取得の途中でデータが追加・削除されても、重複や欠落が起きにくくなります。公式ドキュメントにも、同じ考え方の例(orderBy=OptyId&q=OptyId>1000)があります。ヒントは、この下にあります
  2. 標準の再試行機能に置き換える:requests は、urllib3 の Retry で再試行を設定できます(公式ドキュメントの「Example: Automatic Retries」)。get_json() の再試行を、この下の設定に置き換えて、同じ結果になるか確認してください。再試行を使い切ると、RetryError になります
  3. 並列にページを取る:総件数(totalResults=true)が分かれば、各ページを並列に取れます。ただし、並列度を上げるほど、環境への負荷と 429・503 のリスクが増えます。公式な上限値は今回確認した範囲では公開されていません。2〜3 並列から始めて、どこまで安全か測る方法を考えてください
  4. ログを整える:実行時刻、取得件数、所要時間、再試行の回数を、1 行にまとめて残してください。認証情報や、請求書の中身(個人情報を含みうる項目)は出さないでください
  5. POST の再試行:書き込みを再試行すると、二重登録の恐れがあります。REST 演習2 の発展課題4 を、このスクリプトの設計に当てはめて考えてください

発展課題1 のヒント(キーセット方式)

Section titled “発展課題1 のヒント(キーセット方式)”
ヒントを表示
キーセット方式の fetch_all(差し替え用)
def fetch_all_keyset(session, resource, params, key="InvoiceId"):
last = None
while True:
p = {**params, "limit": PAGE_SIZE, "orderBy": key}
if last is not None:
p["q"] = f"{key} > {last}" # 利用者の q がある場合は、and で結ぶ
page = get_json(session, f"{BASE}/{resource}", p)
yield from page["items"]
if not page["hasMore"] or page["count"] == 0:
return
last = page["items"][-1][key] # fields に key を含めておく

利用者が指定した q(第 2 引数)がある場合は、(利用者の q) and InvoiceId > … のように結びます。このモックでは、30 件すべてを取れます。

発展課題2 のヒント(Retry を使う設定)

Section titled “発展課題2 のヒント(Retry を使う設定)”
ヒントを表示
Retry を使う設定の例
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
retry = Retry(
total=5,
backoff_factor=1,
status_forcelist=[408, 429, 500, 502, 503, 504],
allowed_methods=["GET"],
respect_retry_after_header=True,
)
session.mount("https://", HTTPAdapter(max_retries=retry))

Retry-After があれば、その秒数に従います。ローカルのモック(http://localhost:8788)に向けるときは、mount("http://", …) にします。

項目 このモック 実機で変える・確認する
接続先 https://<このサイトのURL>/api/mock/… https://<your-pod>.fa.<region>.oraclecloud.com/fscmRestApi/resources/11.13.18.05。API_BASE を変える
認証 Basic(何を入れても通る) 環境の認証方式で統合ユーザーを認証する。Basic のほか OAuth 2.0・JWT がある(認証・認可)。トークンは、期限切れ(401)に備えて取り直す
ページサイズ 上限 500(モックの規則) 調達の Integration Playbook は、調達の REST API は 1 回に 500 件を超えて返さないと説明している。財務の invoices の上限は、公式の該当ページに記載がない。応答の limit と count を基準に進める
429・Retry-After X-Mock-Fault で返す 公式の Status Codes には、429 も Retry-After も載っていない。実機で返ってきたら従う、という実装にしておく
再試行する状態 408・429・500・502・503・504 公式の Integration Playbook(調達)は、500・502・503・504 を、タイムアウト処理と指数バックオフで再試行するよう勧めている。503 で応答に「Maintenance」とあるときは、合計で最大 8 時間程度の遅延も考える
応答時間 数十ミリ秒(遅れは X-Mock-Delay で作る) 件数や条件で、数秒以上かかることがある。READ_TIMEOUT を調整し、遅いときは PAGE_SIZE を下げて fields で絞る
データの変化 固定の 30 件 取得中も更新される。offset 方式は重複や欠落が起きうる。キーセット方式や、取得対象を固定する条件(日付など)を検討する
totalResults 正確な件数 公式は推定の行数と説明している。ループの終了は hasMore で判断する
並列度 逐次 公式な上限値は確認できていない。少ない並列数から始める
  • 取得した CSV には、仕入先名や金額など、機密の情報が含まれます。保存先のアクセス権を絞り、不要になったら削除してください
  • 練習用 API で動いたスクリプトが、実機でそのまま動くとは限りません。属性名は describe で、q の構文は公式ドキュメントで確認してください

本サイトは学習目的で個人が作成した非公式の教材であり、Oracle Corporation およびその関連会社とは関係ありません。内容の正確性は保証しません。最新の仕様・試験範囲は必ず公式ドキュメントと Oracle University の情報で確認してください。

模擬テストの問題は公開されている出題範囲(Exam Topics)と公式ドキュメントをもとに独自に作成したもので、実際の試験問題ではありません。

Oracle、Java、MySQL は Oracle Corporation およびその子会社、関連会社の米国およびその他の国における登録商標です。その他の製品名・会社名は各社の商標です。

© 2026 Fusion ERP 実装ラボ