From 0188cbe0bf9550fca6cfb53d4373ed3a1cdf7898 Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Fri, 15 May 2026 15:23:54 +0000 Subject: [PATCH] Show MediaWiki API error messages --- geocode/wikidata.py | 37 +++++++++++++++++++++++++++++++--- lookup.py | 27 ++++++++++++++++++++++--- tests/test_wikidata_api.py | 41 +++++++++++++++++++++++++++++++++++++- 3 files changed, 98 insertions(+), 7 deletions(-) diff --git a/geocode/wikidata.py b/geocode/wikidata.py index 6145050..6af0707 100644 --- a/geocode/wikidata.py +++ b/geocode/wikidata.py @@ -32,6 +32,7 @@ class QueryError(Exception): def __init__(self, query: str, r: requests.Response): """Init.""" + super().__init__(query, r) self.query = query self.r = r @@ -39,12 +40,33 @@ class QueryError(Exception): class APIResponseError(Exception): """Custom exception for API errors with response text.""" - def __init__(self, message: str, response: requests.Response): + def __init__( + self, message: str, response: requests.Response, detail: str | None = None + ): """Init.""" super().__init__(message) + self.message = message + self.detail = detail or message self.response = response +def mediawiki_error_message( + response: requests.Response, json_data: dict[str, typing.Any] | None = None +) -> str: + """Extract the useful error message from a MediaWiki API response.""" + if json_data and isinstance(json_data.get("error"), dict): + error = json_data["error"] + if isinstance(error.get("info"), str): + return typing.cast(str, error["info"]) + if isinstance(error.get("code"), str): + return typing.cast(str, error["code"]) + + if response.text: + return response.text.strip() + + return response.reason or f"HTTP {response.status_code}" + + @backoff.on_exception( backoff.expo, (RequestException, APIResponseError), @@ -58,9 +80,18 @@ def api_call(params: dict[str, str | int]) -> dict[str, typing.Any]: r = requests.get( "https://www.wikidata.org/w/api.php", params=api_params, headers=headers ) - return typing.cast(dict[str, typing.Any], r.json()) + json_data = typing.cast(dict[str, typing.Any], r.json()) except JSONDecodeError: - raise APIResponseError("Failed to decode JSON", r) + detail = mediawiki_error_message(r) + message = f"Wikidata API error (HTTP {r.status_code}): {detail}" + raise APIResponseError(message, r, detail) + + if not r.ok or "error" in json_data: + detail = mediawiki_error_message(r, json_data) + message = f"Wikidata API error (HTTP {r.status_code}): {detail}" + raise APIResponseError(message, r, detail) + + return json_data def get_entity(qid: str) -> dict[str, typing.Any] | None: diff --git a/lookup.py b/lookup.py index bebe7a3..9531063 100755 --- a/lookup.py +++ b/lookup.py @@ -381,9 +381,26 @@ def index() -> str | Response: except (wikidata.QueryError, wikidata.APIResponseError) as e: r = e.r if isinstance(e, wikidata.QueryError) else e.response if r.status_code == 429: - extra = {"Retry-After": r.headers["Retry-After"]} if "Retry-After" in r.headers else {} - return jsonify(error="Rate limited by Wikidata, please try again later"), 429, extra - return jsonify(error=f"Wikidata query failed (HTTP {r.status_code}), please try again later"), 503 + extra = ( + {"Retry-After": r.headers["Retry-After"]} + if "Retry-After" in r.headers + else {} + ) + error = ( + e.detail + if isinstance(e, wikidata.APIResponseError) + else wikidata.mediawiki_error_message(r) + ) + return jsonify(error=error), 429, extra + if isinstance(e, wikidata.APIResponseError): + return jsonify(error=e.message), 503 + return ( + jsonify( + error=f"Wikidata query failed (HTTP {r.status_code}), " + + "please try again later" + ), + 503, + ) except requests.exceptions.RequestException: return jsonify(error="Could not connect to Wikidata, please try again later"), 503 result.pop("element", None) @@ -457,6 +474,10 @@ def build_detail_page(lat: float, lon: float, needs_commons: bool = True) -> str except wikidata.QueryError as e: query, r = e.args return render_template("query_error.html", lat=lat, lon=lon, query=query, r=r) + except wikidata.APIResponseError as e: + return render_template( + "query_error.html", lat=lat, lon=lon, error=e.message, r=e.response + ) element = reply["result"].pop("element", None) geojson = reply["result"].pop("geojson", None) diff --git a/tests/test_wikidata_api.py b/tests/test_wikidata_api.py index 9877e2a..df630a5 100644 --- a/tests/test_wikidata_api.py +++ b/tests/test_wikidata_api.py @@ -2,7 +2,13 @@ import pytest import pytest_mock import requests import responses -from geocode.wikidata import APIResponseError, QueryError, api_call, wdqs +from geocode.wikidata import ( + APIResponseError, + QueryError, + api_call, + mediawiki_error_message, + wdqs, +) max_tries = 5 @@ -53,6 +59,39 @@ def test_api_call_retries_on_connection_error( assert mocked_sleep.call_count == max_tries - 1 +@responses.activate +def test_api_call_uses_mediawiki_error_message( + mocker: pytest_mock.plugin.MockerFixture, +) -> None: + """Test MediaWiki API error messages are preserved.""" + mocker.patch("time.sleep", return_value=None) + mocker.patch("geocode.mail.send_to_admin") + + responses.add( + responses.GET, + "https://www.wikidata.org/w/api.php", + json={"error": {"code": "ratelimited", "info": "Too many requests"}}, + status=429, + headers={"Retry-After": "10"}, + ) + + with pytest.raises(APIResponseError) as exc_info: + api_call({"action": "wbgetentities", "ids": "Q42"}) + + assert exc_info.value.detail == "Too many requests" + assert str(exc_info.value) == "Wikidata API error (HTTP 429): Too many requests" + + +def test_mediawiki_error_message_falls_back_to_response_text() -> None: + """Test plain-text MediaWiki API errors are preserved.""" + response = requests.Response() + response.status_code = 429 + response.reason = "Too Many Requests" + response._content = b"Please slow down" + + assert mediawiki_error_message(response) == "Please slow down" + + def test_wdqs_retry(mocker: pytest_mock.plugin.MockerFixture) -> None: """Test retry for WDQS API calls.""" # Patch 'time.sleep' to instantly return, effectively skipping the sleep