openapi - OpenAPI.raise_on_http_status allows customizing behaviour

aiopenapi3/errors.py

@@ -189,23 +189,27 @@ def __str__(self):
         {self.missing}>"""
 
 
+@dataclasses.dataclass(repr=False)
 class HTTPStatusIndicatedError(HTTPError):
     """The HTTP Status is 4xx or 5xx"""
 
-    pass
+    status_code: int
+    headers: dict[str, str]
+    data: pydantic.BaseModel
+
+    def __str__(self):
+        return f"""<{self.__class__.__name__} {self.status_code} {self.data} {self.headers}>"""
 
 
 @dataclasses.dataclass(repr=False)
 class HttpClientError(HTTPStatusIndicatedError):
     """response code 4xx"""
 
-    headers: dict[str, str]
-    data: pydantic.BaseModel
+    pass
 
 
 @dataclasses.dataclass(repr=False)
 class HttpServerError(HTTPStatusIndicatedError):
     """response code 5xx"""
 
-    headers: dict[str, str]
-    data: pydantic.BaseModel
+    pass

aiopenapi3/openapi.py

@@ -28,7 +28,7 @@
 from . import v31
 from . import log
 from .request import OperationIndex, HTTP_METHODS
-from .errors import ReferenceResolutionError
+from .errors import ReferenceResolutionError, HttpClientError, HttpServerError
 from .loader import Loader, NullLoader
 from .plugin import Plugin, Plugins
 from .base import RootBase, ReferenceBase, SchemaBase, OperationBase, DiscriminatorBase
@@ -232,7 +232,6 @@ def __init__(
         loader: Optional[Loader] = None,
         plugins: Optional[list[Plugin]] = None,
         use_operation_tags: bool = True,
-        raise_on_error: bool = True,
     ) -> None:
         """
         Creates a new OpenAPI document from a loaded spec file.  This is
@@ -245,7 +244,6 @@ def __init__(
         :param loader: the Loader for the description document(s)
         :param plugins: list of plugins
         :param use_operation_tags: honor tags
-        :param raise_on_error: raise an exception if the http status code indicates error
         """
         self._base_url: yarl.URL = yarl.URL(url)
 
@@ -268,9 +266,12 @@ def __init__(
         Maximum Content-Length in Responses - default to 8 MBytes
         """
 
-        self._raise_on_error = raise_on_error
+        self.raise_on_http_status: list[tuple[type[Exception], tuple[int, int]]] = [
+            (HttpClientError, (400, 499)),
+            (HttpServerError, (500, 599)),
+        ]
         """
-        Raise for http status code 400-599
+        Raise for http status code
         """
 
         self._security: dict[str, tuple[str]] = dict()

aiopenapi3/request.py

@@ -217,13 +217,10 @@ def _build_req(self, session: Union[httpx.Client, httpx.AsyncClient]) -> httpx.R
         )
         return req
 
-    def _raise_on_error(self, result: httpx.Response, headers: dict[str, str], data: Union[pydantic.BaseModel, bytes]):
-        if self.api._raise_on_error is False:
-            return
-        if 500 <= result.status_code <= 599:
-            raise HttpServerError(headers, data)
-        elif 400 <= result.status_code <= 499:
-            raise HttpClientError(headers, data)
+    def _raise_on_http_status(self, status_code: int, headers: dict[str, str], data: Union[pydantic.BaseModel, bytes]):
+        for exc, (start, end) in self.api.raise_on_http_status:
+            if start <= status_code <= end:
+                raise exc(status_code, headers, data)
 
     def request(
         self,

aiopenapi3/v20/glue.py

@@ -350,11 +350,11 @@ def _process_request(self, result: httpx.Response) -> tuple["ResponseHeadersType
                 request=self, operationId=self.operation.operationId, unmarshalled=data
             ).unmarshalled
 
-            self._raise_on_error(result, rheaders, data)
+            self._raise_on_http_status(int(status_code), rheaders, data)
 
             return rheaders, data
         elif self.operation.produces and content_type in self.operation.produces:
-            self._raise_on_error(result, rheaders, ctx.received)
+            self._raise_on_http_status(result.status_code, rheaders, ctx.received)
             return rheaders, ctx.received
         else:
             raise ContentTypeError(

aiopenapi3/v30/glue.py

@@ -577,7 +577,7 @@ def _process_request(self, result: httpx.Response) -> tuple["ResponseHeadersType
                 request=self, operationId=self.operation.operationId, unmarshalled=data
             ).unmarshalled
 
-            self._raise_on_error(result, rheaders, data)
+            self._raise_on_http_status(int(status_code), rheaders, data)
 
             return rheaders, data
         else:
@@ -586,7 +586,7 @@ def _process_request(self, result: httpx.Response) -> tuple["ResponseHeadersType
             e.g. application/octet-stream
             but we can't validate it since it's not json.
             """
-            self._raise_on_error(result, rheaders, ctx.received)
+            self._raise_on_http_status(result.status_code, rheaders, ctx.received)
 
             return rheaders, ctx.received