Add ability to throw exception on failed responses

Author: lee-elenbaas

openapi_python_client/parser/properties/__init__.py

@@ -13,6 +13,7 @@
 
 from itertools import chain
 from typing import Any, ClassVar, Dict, Generic, Iterable, List, Optional, Set, Tuple, TypeVar, Union
+from enum import Enum
 
 import attr
 
@@ -35,6 +36,12 @@
 )
 
 
+class ResponseType(Enum):
+    AUTO    = 'auto'     # Automatically decide whether a response is a failed or success response - every response within the rance [200, 300) will be considered success, others as failed
+    FAILED  = 'failed'   # Explicitly specified failed response
+    SUCCESS = 'success'  # Explicitly specified success response
+
+
 @attr.s(auto_attribs=True, frozen=True)
 class AnyProperty(Property):
     """A property that can be any type (used for empty schemas)"""

openapi_python_client/parser/responses.py

@@ -9,7 +9,7 @@
 from .. import schema as oai
 from ..utils import PythonIdentifier
 from .errors import ParseError, PropertyError
-from .properties import AnyProperty, Property, Schemas, property_from_data
+from .properties import ResponseType, AnyProperty, Property, Schemas, property_from_data
 
 
 @attr.s(auto_attribs=True, frozen=True)
@@ -19,6 +19,7 @@ class Response:
     status_code: HTTPStatus
     prop: Property
     source: str
+    failed_response: bool
     data: object
 
 
@@ -50,6 +51,7 @@ def empty_response(
             description=description,
             example=None,
         ),
+        failed_response=status_code < HTTPStatus.OK or status_code >= HTTPStatus.MULTIPLE_CHOICES,
         source="None",
     )
 
@@ -96,6 +98,15 @@ def response_from_data(
             schemas,
         )
 
+    response_type_val = schema_data['x-response-type'] if 'x-response-type' in schema_data else ResponseType.AUTO.value
+    if response_type_val not in ResponseType._value2member_map_:
+        return ParseError(data=data, detail=f"Unsupported x-response-type: {response_type}"), schemas
+    response_type = ResponseType(response_type_val)
+    if response_type == ResponseType.AUTO:
+        failed_response = status_code < HTTPStatus.OK or status_code >= HTTPStatus.MULTIPLE_CHOICES
+    else:
+        failed_response = response_type == ResponseType.FAILED
+
     prop, schemas = property_from_data(
         name=response_name,
         required=True,
@@ -108,4 +119,4 @@ def response_from_data(
     if isinstance(prop, PropertyError):
         return prop, schemas
 
-    return Response(status_code=status_code, prop=prop, source=source, data=data), schemas
+    return Response(status_code=status_code, prop=prop, source=source, failed_response=failed_response, data=data), schemas

openapi_python_client/templates/client.py.jinja

@@ -16,6 +16,8 @@ class Client:
             but can be set to False for testing purposes.
         raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
             status code that was not documented in the source OpenAPI document.
+        raise_on_failed_status: Whether or not to raise an errors.FailedStatus if the API returns a
+            status code that match a failed response (marked using x-failed-response, by default != 200).
     """
 
     base_url: str
@@ -24,6 +26,7 @@ class Client:
     timeout: float = attr.ib(5.0, kw_only=True)
     verify_ssl: Union[str, bool, ssl.SSLContext] = attr.ib(True, kw_only=True)
     raise_on_unexpected_status: bool = attr.ib(False, kw_only=True)
+    raise_on_failed_status: bool = attr.ib(False, kw_only=True)
 
     def get_headers(self) -> Dict[str, str]:
         """ Get headers to be used in all endpoints """

openapi_python_client/templates/endpoint_macros.py.jinja

@@ -161,6 +161,7 @@ Args:
 {% endif %}
 Raises:
     errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+    errors.FailedStatus: If the server returns a failed status code and Client.raise_on_failed_status is True.
     httpx.TimeoutException: If the request takes longer than Client.timeout.
 
 Returns:

openapi_python_client/templates/endpoint_module.py.jinja

@@ -68,8 +68,16 @@ def _parse_response(*, client: Client, response: httpx.Response) -> Optional[{{
         {% else %}
         {{ response.prop.python_name }} = cast({{ response.prop.get_type_string() }}, {{ response.source }})
         {% endif %}
+        {% if response.failed_response %}
+        if client.raise_on_failed_status:
+            raise errors.FailedStatus(f"Failed status code: {response.status_code}", {{ response.prop.python_name }})
+        {% endif %}
         return {{ response.prop.python_name }}
         {% else %}
+        {% if response.failed_response %}
+        if client.raise_on_failed_status:
+            raise errors.FailedStatus(f"Failed status code: {response.status_code}")
+        {% endif %}
         return None
         {% endif %}
     {% endfor %}

openapi_python_client/templates/errors.py.jinja

@@ -4,4 +4,10 @@ class UnexpectedStatus(Exception):
     """ Raised by api functions when the response status an undocumented status and Client.raise_on_unexpected_status is True """
     ...
 
-__all__ = ["UnexpectedStatus"]
+
+class FailedStatus(Exception):
+    """ Raised by api functions when the response status is a failed status and Client.raise_on_failed_status is True """
+    error: object = None
+
+
+__all__ = ["UnexpectedStatus", "FailedStatus"]