added get_response_json function to return json from aioweb

src/monitor_utils/sessions.py

@@ -171,3 +171,43 @@ async def send_request(
     )
     response.raise_for_status()
     return response
+
+
+def get_response_json(
+    response: Response,
+    *,
+    logger: logging.Logger = LOGGER,
+    expected: type | tuple[type, ...] = dict,
+    **context,
+) -> dict | list | None:
+    """Parse a response body as JSON, or log it and return None.
+
+    aioweb's response.json() yields None for a non-JSON body, so a single
+    isinstance check covers both the empty/invalid case and a body of the
+    wrong shape. Callers stop on None the same way they did with the inline
+    check this replaces.
+
+    Args:
+        response: A 2xx aioweb Response from send_request.
+        logger: Logger the warning is emitted on; pass the monitor's own so
+            the record stays attributed to that monitor.
+        expected: Type(s) the parsed body must be an instance of.
+        **context: Extra fields for the warning record (category, page, ...).
+
+    Returns:
+        The parsed body when it matches `expected`, else None.
+    """
+
+    data = response.json()
+    if isinstance(data, expected):
+        return data
+
+    logger.warning(
+        "unexpected response",
+        extra={
+            **context,
+            "body_type": type(data).__name__,
+            "body": str(data)[:200],
+        },
+    )
+    return None

tests/test_get_response_json.py

@@ -0,0 +1,80 @@
+"""unit tests: sessions.get_response_json parses a body or logs and returns None
+
+run from the repo root:
+    uv run pytest tests/test_response_json.py
+
+offline — the aioweb Response is faked at the leaf; nothing is sent.
+"""
+
+import logging
+
+import pytest
+
+from monitor_utils.sessions import get_response_json
+
+
+class FakeResponse:
+    def __init__(self, body):
+        self.body = body
+
+    def json(self):
+        return self.body
+
+
+def warnings(caplog):
+    return [r for r in caplog.records if r.levelno == logging.WARNING]
+
+
+def test_dict_body_is_returned_without_logging(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse({"a": 1})) == {"a": 1}
+    assert warnings(caplog) == []
+
+
+def test_none_body_logs_and_returns_none(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse(None), category="c1", page=2) is None
+    (record,) = warnings(caplog)
+    assert record.message == "unexpected response"
+    assert record.body_type == "NoneType"
+    assert record.body == "None"
+    assert (record.category, record.page) == ("c1", 2)
+
+
+@pytest.mark.parametrize("body", [[1, 2], "text", 42, b"bytes"])
+def test_wrong_shape_logs_type_and_truncated_body(body, caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse(body)) is None
+    (record,) = warnings(caplog)
+    assert record.body_type == type(body).__name__
+    assert record.body == str(body)[:200]
+
+
+def test_body_preview_is_capped_at_200_chars(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        get_response_json(FakeResponse("x" * 1000))
+    (record,) = warnings(caplog)
+    assert len(record.body) == 200
+
+
+def test_expected_accepts_other_types(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse([1]), expected=list) == [1]
+        assert get_response_json(FakeResponse([1]), expected=(dict, list)) == [1]
+        assert get_response_json(FakeResponse({}), expected=list) is None
+    assert len(warnings(caplog)) == 1
+
+
+def test_warning_goes_to_the_passed_logger(caplog):
+    with caplog.at_level(logging.WARNING):
+        get_response_json(
+            FakeResponse(None), logger=logging.getLogger("woot.woot_monitor")
+        )
+    (record,) = warnings(caplog)
+    assert record.name == "woot.woot_monitor"
+
+
+def test_empty_dict_counts_as_a_dict(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse({})) == {}
+    assert warnings(caplog) == []