Use custom response type in exceptions, to avoid needing to use requests' API to handle exceptions, and to avoid relying on requests's types which are incomplete

Author: adamtheturtle

Makefile

@@ -17,6 +17,7 @@ lint: \
     pip-missing-reqs \
     pyproject-fmt \
     pyright \
+    pyright-verifytypes \
     pyroma \
     spelling \
     vulture \

docs/source/api-reference.rst

@@ -5,6 +5,10 @@ API Reference
    :undoc-members:
    :members:
 
+.. automodule:: vws.exceptions.response
+   :undoc-members:
+   :members:
+
 .. automodule:: vws.reports
    :undoc-members:
    :members:

lint.mk

@@ -48,6 +48,10 @@ pyroma:
 pyright:
 	pyright .
 
+.PHONY: pyright-verifytypes
+pyright-verifytypes:
+	pyright --verifytypes vws
+
 .PHONY: vulture
 vulture:
 	vulture --min-confidence 100 --exclude _vendor --exclude .eggs .

src/vws/exceptions/base_exceptions.py

@@ -5,10 +5,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from requests import Response
+from .response import Response
 
 
 class CloudRecoException(Exception):

src/vws/exceptions/custom_exceptions.py

@@ -5,7 +5,7 @@
 """
 
 
-from requests import Response
+from .response import Response
 
 
 class OopsAnErrorOccurredPossiblyBadName(Exception):

src/vws/exceptions/response.py

@@ -0,0 +1,16 @@
+"""Responses for exceptions."""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class Response:
+    """
+    A response from a request.
+    """
+
+    text: str
+    url: str
+    status_code: int
+    headers: dict[str, str]
+    request_body: bytes | str | None

src/vws/exceptions/vws_exceptions.py

@@ -137,8 +137,8 @@ def target_name(self) -> str:
         """
         The target name which already exists.
         """
-        response_body = self.response.request.body or b""
-        request_json = json.loads(response_body)
+        response_body = self.response.request_body or b""
+        request_json = json.loads(s=response_body)
         return str(request_json["name"])
 
 

src/vws/query.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import datetime
+import json
 from http import HTTPStatus
 from typing import Any, BinaryIO
 from urllib.parse import urljoin
@@ -24,6 +25,7 @@
     RequestEntityTooLarge,
     ServerError,
 )
+from vws.exceptions.response import Response
 from vws.include_target_data import CloudRecoIncludeTargetData
 from vws.reports import QueryResult, TargetData
 
@@ -133,14 +135,21 @@ def query(
             "Content-Type": content_type_header,
         }
 
-        response = requests.request(
+        requests_response = requests.request(
             method=method,
             url=urljoin(base=self._base_vwq_url, url=request_path),
             headers=headers,
             data=content,
             # We should make the timeout customizable.
             timeout=None,
         )
+        response = Response(
+            text=requests_response.text,
+            url=requests_response.url,
+            status_code=requests_response.status_code,
+            headers=dict(requests_response.headers),
+            request_body=requests_response.request.body,
+        )
 
         if response.status_code == HTTPStatus.REQUEST_ENTITY_TOO_LARGE:
             raise RequestEntityTooLarge
@@ -153,7 +162,7 @@ def query(
         ):  # pragma: no cover
             raise ServerError(response=response)
 
-        result_code = response.json()["result_code"]
+        result_code = json.loads(s=response.text)["result_code"]
         if result_code != "Success":
             exception = {
                 "AuthenticationFailure": AuthenticationFailure,
@@ -164,7 +173,7 @@ def query(
             raise exception(response=response)
 
         result: list[QueryResult] = []
-        result_list = list(response.json()["results"])
+        result_list = list(json.loads(s=response.text)["results"])
         for item in result_list:
             target_data: TargetData | None = None
             if "target_data" in item:

src/vws/vws.py

@@ -13,7 +13,6 @@
 from urllib.parse import urljoin
 
 import requests
-from requests import Response
 from vws_auth_tools import authorization_header, rfc_1123_date
 
 from vws.exceptions.custom_exceptions import (
@@ -48,6 +47,8 @@
     TargetSummaryReport,
 )
 
+from .exceptions.response import Response
+
 if TYPE_CHECKING:
     import io
 
@@ -107,7 +108,7 @@ def _target_api_request(
 
     url = urljoin(base=base_vws_url, url=request_path)
 
-    return requests.request(
+    requests_response = requests.request(
         method=method,
         url=url,
         headers=headers,
@@ -116,6 +117,14 @@ def _target_api_request(
         timeout=None,
     )
 
+    return Response(
+        text=requests_response.text,
+        url=requests_response.url,
+        status_code=requests_response.status_code,
+        headers=dict(requests_response.headers),
+        request_body=requests_response.request.body,
+    )
+
 
 class VWS:
     """
@@ -198,7 +207,7 @@ def _make_request(
         ):  # pragma: no cover
             raise ServerError(response=response)
 
-        result_code = response.json()["result_code"]
+        result_code = json.loads(s=response.text)["result_code"]
 
         if result_code == expected_result_code:
             return response
@@ -303,7 +312,7 @@ def add_target(
             expected_result_code="TargetCreated",
         )
 
-        return str(response.json()["target_id"])
+        return str(json.loads(response.text)["target_id"])
 
     def get_target_record(self, target_id: str) -> TargetStatusAndRecord:
         """
@@ -340,7 +349,7 @@ def get_target_record(self, target_id: str) -> TargetStatusAndRecord:
             expected_result_code="Success",
         )
 
-        result_data = response.json()
+        result_data = json.loads(s=response.text)
         status = TargetStatuses(result_data["status"])
         target_record_dict = dict(result_data["target_record"])
         target_record = TargetRecord(
@@ -436,7 +445,7 @@ def list_targets(self) -> list[str]:
             expected_result_code="Success",
         )
 
-        return list(response.json()["results"])
+        return list(json.loads(response.text)["results"])
 
     def get_target_summary_report(self, target_id: str) -> TargetSummaryReport:
         """
@@ -473,7 +482,7 @@ def get_target_summary_report(self, target_id: str) -> TargetSummaryReport:
             expected_result_code="Success",
         )
 
-        result_data = dict(response.json())
+        result_data = dict(json.loads(response.text))
         return TargetSummaryReport(
             status=TargetStatuses(result_data["status"]),
             database_name=result_data["database_name"],
@@ -516,7 +525,7 @@ def get_database_summary_report(self) -> DatabaseSummaryReport:
             expected_result_code="Success",
         )
 
-        response_data = dict(response.json())
+        response_data = dict(json.loads(response.text))
         return DatabaseSummaryReport(
             active_images=response_data["active_images"],
             current_month_recos=response_data["current_month_recos"],
@@ -603,7 +612,7 @@ def get_duplicate_targets(self, target_id: str) -> list[str]:
             expected_result_code="Success",
         )
 
-        return list(response.json()["similar_targets"])
+        return list(json.loads(s=response.text)["similar_targets"])
 
     def update_target(
         self,