【Python】Requestsで通信の失敗を扱う：timeout・HTTPエラー・JSON解析

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
import time
import requests

class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass
    def do_GET(self):
        if self.path == "/disconnect":
            self.close_connection = True
            return
        if self.path == "/slow":
            time.sleep(0.2)
        status = 503 if self.path == "/status" else 200
        body = b'not json' if self.path == "/json" else b'{"value": 3}'
        if self.path == "/schema":
            body = b'{"value": "wrong"}'
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        try:
            self.wfile.write(body)
        except (BrokenPipeError, ConnectionResetError):
            pass  # The timeout client has already disconnected.

def fetch_value(session, url):
    with session.get(url, timeout=(1, 0.05), allow_redirects=False) as response:
        response.raise_for_status()
        if response.status_code != 200:
            raise ValueError("expected status 200")
        data = response.json()
    if not isinstance(data, dict) or type(data.get("value")) is not int:
        raise ValueError("value must be an integer")
    return data["value"]

server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
server.daemon_threads = False  # server_close waits for these short fixture handlers.
thread = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01})
thread.start()
base = f"http://127.0.0.1:{server.server_port}"
try:
    with requests.Session() as session:
        session.trust_env = False  # Keep this fixture independent of proxy settings.
        assert fetch_value(session, base + "/ok") == 3
        print("ok: 3")
        expected = {"status": requests.HTTPError,
                    "json": requests.exceptions.JSONDecodeError,
                    "slow": requests.ReadTimeout,
                    "schema": ValueError,
                    "disconnect": requests.ConnectionError}
        for name, error_type in expected.items():
            try:
                fetch_value(session, base + "/" + name)
            except error_type as exc:
                print(name + ":", type(exc).__name__)
            else:
                raise AssertionError(name)
finally:
    server.shutdown()
    server.server_close()
    thread.join(timeout=2)
assert not thread.is_alive()
print("loopback server stopped: True")