v20/3x - raise on http status 4xx and 5xx

customizable, docs & tests

Author: commonism

aiopenapi3/errors.py

@@ -3,7 +3,7 @@
 import dataclasses
 
 import httpx
-
+import pydantic
 
 if typing.TYPE_CHECKING:
     from ._types import (
@@ -187,3 +187,29 @@ class HeadersMissingError(ResponseError):
     def __str__(self):
         return f"""<{self.__class__.__name__} {self.response.request.method} '{self.response.request.url.path}' ({self.operation.operationId})
         {self.missing}>"""
+
+
+@dataclasses.dataclass(repr=False)
+class HTTPStatusIndicatedError(HTTPError):
+    """The HTTP Status is 4xx or 5xx"""
+
+    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"""
+
+    pass
+
+
+@dataclasses.dataclass(repr=False)
+class HTTPServerError(HTTPStatusIndicatedError):
+    """response code 5xx"""
+
+    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
@@ -266,6 +266,14 @@ def __init__(
         Maximum Content-Length in Responses - default to 8 MBytes
         """
 
+        self.raise_on_http_status: list[tuple[type[Exception], tuple[int, int]]] = [
+            (HTTPClientError, (400, 499)),
+            (HTTPServerError, (500, 599)),
+        ]
+        """
+        Raise for http status code
+        """
+
         self._security: dict[str, tuple[str]] = dict()
         """
         authorization informations

aiopenapi3/request.py

@@ -27,8 +27,7 @@ async def aclosing(thing):
 
 from .base import HTTP_METHODS, ReferenceBase
 from .version import __version__
-from .errors import RequestError, OperationIdDuplicationError
-
+from .errors import RequestError, OperationIdDuplicationError, HTTPServerError, HTTPClientError
 
 if typing.TYPE_CHECKING:
     from ._types import (
@@ -218,6 +217,11 @@ def _build_req(self, session: Union[httpx.Client, httpx.AsyncClient]) -> httpx.R
         )
         return req
 
+    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,
         data: Optional["RequestData"] = None,

aiopenapi3/v20/glue.py

@@ -17,6 +17,7 @@
 from ..request import RequestBase, AsyncRequestBase
 from ..errors import HTTPStatusError, ContentTypeError, ResponseSchemaError, ResponseDecodingError, HeadersMissingError
 
+
 from .parameter import Parameter
 from .root import Root
 
@@ -348,9 +349,13 @@ def _process_request(self, result: httpx.Response) -> tuple["ResponseHeadersType
             data = self.api.plugins.message.unmarshalled(
                 request=self, operationId=self.operation.operationId, unmarshalled=data
             ).unmarshalled
+
+            self._raise_on_http_status(int(status_code), rheaders, data)
+
             return rheaders, data
         elif self.operation.produces and content_type in self.operation.produces:
-            return rheaders, result.content
+            self._raise_on_http_status(result.status_code, rheaders, ctx.received)
+            return rheaders, ctx.received
         else:
             raise ContentTypeError(
                 self.operation,

aiopenapi3/v30/glue.py

@@ -576,13 +576,18 @@ def _process_request(self, result: httpx.Response) -> tuple["ResponseHeadersType
             data = self.api.plugins.message.unmarshalled(
                 request=self, operationId=self.operation.operationId, unmarshalled=data
             ).unmarshalled
+
+            self._raise_on_http_status(int(status_code), rheaders, data)
+
             return rheaders, data
         else:
             """
             We have received a valid (i.e. expected) content type,
             e.g. application/octet-stream
             but we can't validate it since it's not json.
             """
+            self._raise_on_http_status(result.status_code, rheaders, ctx.received)
+
             return rheaders, ctx.received
 
 

docs/source/api.rst

@@ -7,7 +7,7 @@ API
 General
 =======
 .. autoclass:: aiopenapi3.OpenAPI
-    :members: authenticate, createRequest, load_async, load_file, load_sync, loads, clone, cache_load, cache_store, _
+    :members: authenticate, createRequest, load_async, load_file, load_sync, loads, clone, cache_load, cache_store, _, raise_on_http_status
 
 
 Requests
@@ -266,7 +266,7 @@ Exceptions
 
 There is different types of Exceptions used depending on the subsystem/failure.
 
-.. inheritance-diagram:: aiopenapi3.errors.SpecError aiopenapi3.errors.ReferenceResolutionError aiopenapi3.errors.OperationParameterValidationError aiopenapi3.errors.ParameterFormatError aiopenapi3.errors.HTTPError aiopenapi3.errors.RequestError aiopenapi3.errors.ResponseError aiopenapi3.errors.ContentTypeError aiopenapi3.errors.HTTPStatusError aiopenapi3.errors.ResponseDecodingError aiopenapi3.errors.ResponseSchemaError aiopenapi3.errors.ContentLengthExceededError aiopenapi3.errors.HeadersMissingError
+.. inheritance-diagram:: aiopenapi3.errors.SpecError aiopenapi3.errors.ReferenceResolutionError aiopenapi3.errors.OperationParameterValidationError aiopenapi3.errors.ParameterFormatError aiopenapi3.errors.HTTPError aiopenapi3.errors.RequestError aiopenapi3.errors.ResponseError aiopenapi3.errors.ContentTypeError aiopenapi3.errors.HTTPStatusError aiopenapi3.errors.ResponseDecodingError aiopenapi3.errors.ResponseSchemaError aiopenapi3.errors.ContentLengthExceededError aiopenapi3.errors.HeadersMissingError aiopenapi3.errors.HTTPStatusIndicatedError aiopenapi3.errors.HTTPClientError aiopenapi3.errors.HTTPServerError
     :top-classes: aiopenapi3.errors.BaseError
     :parts: -2
 
@@ -351,6 +351,24 @@ document.
     :members:
     :undoc-members:
 
+HTTP Status
+-----------
+.. inheritance-diagram:: aiopenapi3.errors.HTTPStatusIndicatedError aiopenapi3.errors.HTTPClientError aiopenapi3.errors.HTTPServerError
+   :top-classes: aiopenapi3.errors.HTTPStatusIndicatedError
+   :parts: -2
+
+.. autoexception:: HTTPStatusIndicatedError
+    :members:
+    :undoc-members:
+
+.. autoexception:: HTTPClientError
+    :members:
+    :undoc-members:
+
+.. autoexception:: HTTPServerError
+    :members:
+    :undoc-members:
+
 Extra
 =====
 

docs/source/use.rst

@@ -188,6 +188,14 @@ And …
 
     api._.repoDelete(parameters={"owner":user.login, "repo":"rtd"})
 
+
+Request Errors
+--------------
+
+Starting version 0.8.0 aiopenapi3 :ref:`raises <api:HTTP Status>` HTTPClientError for 400 <= http_code <= 499 and HTTPServerError for 500 to 599.
+
+This is customizable via :obj:`aiopenapi3.OpenAPI.raise_on_http_status`.
+
 async
 =====
 Difference when using asyncio - await.

pyproject.toml

@@ -107,6 +107,7 @@ tests = [
     "pytest-asyncio>=0.24.0",
     "pytest-httpx",
     "pytest-cov",
+    "pytest-mock",
     "fastapi",
     "fastapi-versioning",
     "uvloop == 0.21.0b1; python_version >= '3.13'",

tests/api/v2/main.py

@@ -12,6 +12,8 @@
 
 from fastapi_versioning import versioned_api_route
 
+from pydantic import RootModel
+
 router = APIRouter(route_class=versioned_api_route(2))
 
 ZOO = dict()
@@ -57,9 +59,9 @@ def listPet(limit: Optional[int] = None) -> schema.Pets:
 
 @router.get("/pets/{petId}", operation_id="getPet", response_model=schema.Pet, responses={404: {"model": schema.Error}})
 def getPet(pet_id: str = Path(..., alias="petId")) -> schema.Pets:
-    for k, v in ZOO.items():
-        if pet_id == v.identifier:
-            return v
+    for k, pet in ZOO.items():
+        if pet_id == pet.identifier:
+            return pet
     else:
         return JSONResponse(
             status_code=starlette.status.HTTP_404_NOT_FOUND,
@@ -73,10 +75,10 @@ def getPet(pet_id: str = Path(..., alias="petId")) -> schema.Pets:
 def deletePet(
     response: Response,
     x_raise_nonexist: Annotated[Union[bool, None], Header()],
-    pet_id: uuid.UUID = Path(..., alias="petId"),
+    pet_id: str = Path(..., alias="petId"),
 ) -> None:
-    for k, v in ZOO.items():
-        if pet_id == v.identifier:
+    for k, pet in ZOO.items():
+        if pet_id == pet.identifier:
             del ZOO[k]
             response.status_code = starlette.status.HTTP_204_NO_CONTENT
             return response

tests/apiv1_test.py

@@ -55,8 +55,9 @@ async def test_createPet(server, client):
     assert type(r).model_json_schema() == client.components.schemas["Pet"].get_type().model_json_schema()
     assert h["X-Limit-Remain"] == 5
 
-    r = await asyncio.to_thread(client._.createPet, data={"pet": {"name": r.name}})
-    assert type(r).model_json_schema() == client.components.schemas["Error"].get_type().model_json_schema()
+    with pytest.raises(aiopenapi3.errors.HTTPClientError) as e:
+        await asyncio.to_thread(client._.createPet, data={"pet": {"name": r.name}})
+    assert type(e.value.data).model_json_schema() == client.components.schemas["Error"].get_type().model_json_schema()
 
 
 @pytest.mark.asyncio(loop_scope="session")
@@ -74,15 +75,18 @@ async def test_getPet(server, client):
     # assert type(r).model_json_schema() == type(pet).model_json_schema()
     assert r.id == pet.id
 
-    r = await asyncio.to_thread(client._.getPet, parameters={"petId": -1})
-    assert type(r).model_json_schema() == client.components.schemas["Error"].get_type().model_json_schema()
+    with pytest.raises(aiopenapi3.errors.HTTPClientError) as e:
+        await asyncio.to_thread(client._.getPet, parameters={"petId": -1})
+
+    assert type(e.value.data).model_json_schema() == client.components.schemas["Error"].get_type().model_json_schema()
 
 
 @pytest.mark.asyncio(loop_scope="session")
 async def test_deletePet(server, client):
-    r = await asyncio.to_thread(client._.deletePet, parameters={"petId": -1})
-    print(r)
-    assert type(r).model_json_schema() == client.components.schemas["Error"].get_type().model_json_schema()
+    with pytest.raises(aiopenapi3.errors.HTTPClientError) as e:
+        await asyncio.to_thread(client._.deletePet, parameters={"petId": -1})
+
+    assert type(e.value.data).model_json_schema() == client.components.schemas["Error"].get_type().model_json_schema()
 
     await asyncio.to_thread(client._.createPet, **randomPet(uuid.uuid4()))
     zoo = await asyncio.to_thread(client._.listPet)