到達点は、通信期限、ステータス、本文、型を順に確認することです。前提は第07・10回です。
ネットワークには部分的な失敗がある
接続できない場合、接続できたがエラー応答の場合、成功応答だが本文が不正な場合は別です。すべてを「データなし」と扱うと障害に気付けません。次の例は外部サービスなしで、ローカルのJSON応答を確認します。
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from urllib.request import urlopen
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
body = json.dumps({"minutes": 25}).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, format, *args):
pass
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
url = f"http://127.0.0.1:{server.server_port}/study"
with urlopen(url, timeout=3) as response:
if response.status != 200:
raise ValueError("unexpected status")
if response.headers.get_content_type() != "application/json":
raise ValueError("unexpected content type")
body = response.read(1025)
if len(body) > 1024:
raise ValueError("response too large")
data = json.loads(body.decode("utf-8"))
if not isinstance(data, dict) or type(data.get("minutes")) is not int:
raise ValueError("invalid response structure")
if data["minutes"] < 0:
raise ValueError("invalid minutes")
print(data["minutes"])
finally:
server.shutdown()
server.server_close()
thread.join()
main.pyの出力は25です。http.serverはここでは教材用で、本番のAPIサーバーとしての設計を示していません。urllibのtimeoutはソケット操作の待ち時間に関わる値で、厳密な処理全体の締切と同じではありません。
APIを使うときの契約
本物の通信ではHTTPSを使います。認証トークンは環境や設定から読み、ログやURLへ混ぜません。401は認証、403は権限、429は頻度制限など、状況に応じて扱います。再試行するなら上限回数と待ち時間を決めます。更新要求は応答だけ失われた可能性があるので、同じ要求を二度処理しない設計が必要です。
練習と解答
練習:サーバーがminutesをTrueで返した場合に、isinstanceだけでは見落とす理由を説明します。
解答:boolはintのサブクラスなので isinstance(True, int) はTrueです。今回の契約では真偽値を時間と認めないため、型の完全一致を使っています。外部入力の検証は、言語の型階層とデータ仕様の両方を考えます。
公式資料
urllib.requestが参照先です。
Python全20回の目次 | 前の回 | 次の回