feat(parser): #1271 Support simple patterned http status codes

Author: obs-gh-peterkolloch

end_to_end_tests/__snapshots__/test_end_to_end.ambr

@@ -6,7 +6,7 @@
 
   WARNING parsing GET / within default.
 
-  Invalid response status code abcdef (not a valid HTTP status code), response will be omitted from generated client
+  Invalid response status code pattern: abcdef, response will be omitted from generated client
 
 
   If you believe this was a mistake or this tool is missing a feature you need, please open an issue at https://github.com/openapi-generators/openapi-python-client/issues/new/choose

openapi_python_client/parser/openapi.py

@@ -2,7 +2,6 @@
 from collections.abc import Iterator
 from copy import deepcopy
 from dataclasses import dataclass, field
-from http import HTTPStatus
 from typing import Any, Optional, Protocol, Union
 
 from pydantic import ValidationError
@@ -26,7 +25,7 @@
     property_from_data,
 )
 from .properties.schemas import parameter_from_reference
-from .responses import Response, response_from_data
+from .responses import HTTPStatusPattern, Response, response_from_data
 
 _PATH_PARAM_REGEX = re.compile("{([a-zA-Z_-][a-zA-Z0-9_-]*)}")
 
@@ -162,19 +161,9 @@ def _add_responses(
     ) -> tuple["Endpoint", Schemas]:
         endpoint = deepcopy(endpoint)
         for code, response_data in data.items():
-            status_code: HTTPStatus
-            try:
-                status_code = HTTPStatus(int(code))
-            except ValueError:
-                endpoint.errors.append(
-                    ParseError(
-                        detail=(
-                            f"Invalid response status code {code} (not a valid HTTP "
-                            f"status code), response will be omitted from generated "
-                            f"client"
-                        )
-                    )
-                )
+            status_code = HTTPStatusPattern.parse(code)
+            if isinstance(status_code, ParseError):
+                endpoint.errors.append(status_code)
                 continue
 
             response, schemas = response_from_data(
@@ -190,7 +179,7 @@ def _add_responses(
                 endpoint.errors.append(
                     ParseError(
                         detail=(
-                            f"Cannot parse response for status code {status_code}{detail_suffix}, "
+                            f"Cannot parse response for status code {code}{detail_suffix}, "
                             f"response will be omitted from generated client"
                         ),
                         data=response.data,
@@ -202,6 +191,7 @@ def _add_responses(
             endpoint.relative_imports |= response.prop.get_lazy_imports(prefix=models_relative_prefix)
             endpoint.relative_imports |= response.prop.get_imports(prefix=models_relative_prefix)
             endpoint.responses.append(response)
+        endpoint.responses.sort()
         return endpoint, schemas
 
     @staticmethod

openapi_python_client/parser/responses.py

@@ -1,6 +1,5 @@
-__all__ = ["Response", "response_from_data"]
+__all__ = ["HTTPStatusPattern", "Response", "response_from_data"]
 
-from http import HTTPStatus
 from typing import Optional, TypedDict, Union
 
 from attrs import define
@@ -28,15 +27,85 @@ class _ResponseSource(TypedDict):
 NONE_SOURCE = _ResponseSource(attribute="None", return_type="None")
 
 
-@define
+class HTTPStatusPattern:
+    """Status code patterns come in three flavors, in order of precedence:
+    1. Specific status codes, such as 200. This is represented by `min` and `max` being the same.
+    2. Ranges of status codes, such as 2XX. This is represented by `min` and `max` being different.
+    3. The special `default` status code, which is used when no other status codes match. `range` is `None` in this case.
+
+    https://github.com/openapi-generators/openapi-python-client/blob/61b6c54994e2a6285bb422ee3b864c45b5d88c15/openapi_python_client/schema/3.1.0.md#responses-object
+    """
+
+    pattern: str
+    range: Optional[tuple[int, int]]
+
+    def __init__(self, *, pattern: str, code_range: Optional[tuple[int, int]]):
+        """Initialize with a range of status codes or None for the default case."""
+        self.pattern = pattern
+        self.range = code_range
+
+    @staticmethod
+    def parse(pattern: str) -> Union["HTTPStatusPattern", ParseError]:
+        """Parse a status code pattern such as 2XX or 404"""
+        if pattern == "default":
+            return HTTPStatusPattern(pattern=pattern, code_range=None)
+
+        if pattern.endswith("XX") and pattern[0].isdigit():
+            first_digit = int(pattern[0])
+            return HTTPStatusPattern(pattern=pattern, code_range=(first_digit * 100, first_digit * 100 + 99))
+
+        try:
+            code = int(pattern)
+            return HTTPStatusPattern(pattern=pattern, code_range=(code, code))
+        except ValueError:
+            return ParseError(
+                detail=(
+                    f"Invalid response status code pattern: {pattern}, response will be omitted from generated client"
+                )
+            )
+
+    def is_range(self) -> bool:
+        """Check if this is a range of status codes, such as 2XX"""
+        return self.range is not None and self.range[0] != self.range[1]
+
+    def __lt__(self, other: "HTTPStatusPattern") -> bool:
+        """Compare two HTTPStatusPattern objects based on the order they should be applied in"""
+        if self.range is None:
+            return False  # Default gets applied last
+        if other.range is None:
+            return True  # Other is default, so this one gets applied first
+
+        # Specific codes appear before ranges
+        if self.is_range() and not other.is_range():
+            return False
+        if not self.is_range() and other.is_range():
+            return True
+
+        # Order specific codes numerically
+        return self.range[0] < other.range[0]
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, HTTPStatusPattern):
+            return False
+        return self.range == other.range
+
+    def __hash__(self) -> int:
+        return hash(self.range)
+
+
+@define(order=False)
 class Response:
     """Describes a single response for an endpoint"""
 
-    status_code: HTTPStatus
+    status_code: HTTPStatusPattern
     prop: Property
     source: _ResponseSource
     data: Union[oai.Response, oai.Reference]  # Original data which created this response, useful for custom templates
 
+    def __lt__(self, other: "Response") -> bool:
+        """Compare two responses based on the order in which they should be applied in"""
+        return self.status_code < other.status_code
+
 
 def _source_by_content_type(content_type: str, config: Config) -> Optional[_ResponseSource]:
     parsed_content_type = utils.get_content_type(content_type, config)
@@ -59,7 +128,7 @@ def _source_by_content_type(content_type: str, config: Config) -> Optional[_Resp
 
 def empty_response(
     *,
-    status_code: HTTPStatus,
+    status_code: HTTPStatusPattern,
     response_name: str,
     config: Config,
     data: Union[oai.Response, oai.Reference],
@@ -82,7 +151,7 @@ def empty_response(
 
 def response_from_data(  # noqa: PLR0911
     *,
-    status_code: HTTPStatus,
+    status_code: HTTPStatusPattern,
     data: Union[oai.Response, oai.Reference],
     schemas: Schemas,
     responses: dict[str, Union[oai.Response, oai.Reference]],
@@ -91,7 +160,7 @@ def response_from_data(  # noqa: PLR0911
 ) -> tuple[Union[Response, ParseError], Schemas]:
     """Generate a Response from the OpenAPI dictionary representation of it"""
 
-    response_name = f"response_{status_code}"
+    response_name = f"response_{status_code.pattern}"
     if isinstance(data, oai.Reference):
         ref_path = parse_reference_path(data.ref)
         if isinstance(ref_path, ParseError):

openapi_python_client/templates/endpoint_module.py.jinja

@@ -66,8 +66,16 @@ def _get_kwargs(
 
 
 def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[{{ return_string }}]:
+    {% set has_default = False %}
     {% for response in endpoint.responses %}
-    if response.status_code == {{ response.status_code.value }}:
+    {% set code_range = response.status_code.range %}
+    {% if code_range == None %} {% set has_default = True %}
+    else:
+    {% elif code_range[0] == code_range[1] %}
+    if response.status_code == {{ code_range[0] }}:
+    {% else %}
+    if {{ code_range[0] }} <= response.status_code <= {{ code_range[1] }}:
+    {% endif %}
         {% if parsed_responses %}{% import "property_templates/" + response.prop.template as prop_template %}
         {% if prop_template.construct %}
         {{ prop_template.construct(response.prop, response.source.attribute) | indent(8) }}
@@ -81,10 +89,12 @@ def _parse_response(*, client: Union[AuthenticatedClient, Client], response: htt
         return None
         {% endif %}
     {% endfor %}
+    {% if not has_default %}
     if client.raise_on_unexpected_status:
         raise errors.UnexpectedStatus(response.status_code, response.content)
     else:
         return None
+    {% endif %}
 
 
 def _build_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Response[{{ return_string }}]:

tests/test_parser/test_openapi.py

@@ -24,80 +24,6 @@ def make_endpoint(self):
             relative_imports={"import_3"},
         )
 
-    @pytest.mark.parametrize("response_status_code", ["not_a_number", 499])
-    def test__add_responses_status_code_error(self, response_status_code, mocker):
-        schemas = Schemas()
-        response_1_data = mocker.MagicMock()
-        data = {
-            response_status_code: response_1_data,
-        }
-        endpoint = self.make_endpoint()
-        parse_error = ParseError(data=mocker.MagicMock())
-        response_from_data = mocker.patch(f"{MODULE_NAME}.response_from_data", return_value=(parse_error, schemas))
-        config = MagicMock()
-
-        response, schemas = Endpoint._add_responses(
-            endpoint=endpoint, data=data, schemas=schemas, responses={}, config=config
-        )
-
-        assert response.errors == [
-            ParseError(
-                detail=f"Invalid response status code {response_status_code} (not a valid HTTP status code), "
-                "response will be omitted from generated client"
-            )
-        ]
-        response_from_data.assert_not_called()
-
-    def test__add_responses_error(self, mocker):
-        schemas = Schemas()
-        response_1_data = mocker.MagicMock()
-        response_2_data = mocker.MagicMock()
-        data = {
-            "200": response_1_data,
-            "404": response_2_data,
-        }
-        endpoint = self.make_endpoint()
-        parse_error = ParseError(data=mocker.MagicMock(), detail="some problem")
-        response_from_data = mocker.patch(f"{MODULE_NAME}.response_from_data", return_value=(parse_error, schemas))
-        config = MagicMock()
-
-        response, schemas = Endpoint._add_responses(
-            endpoint=endpoint, data=data, schemas=schemas, responses={}, config=config
-        )
-
-        response_from_data.assert_has_calls(
-            [
-                mocker.call(
-                    status_code=200,
-                    data=response_1_data,
-                    schemas=schemas,
-                    responses={},
-                    parent_name="name",
-                    config=config,
-                ),
-                mocker.call(
-                    status_code=404,
-                    data=response_2_data,
-                    schemas=schemas,
-                    responses={},
-                    parent_name="name",
-                    config=config,
-                ),
-            ]
-        )
-        assert response.errors == [
-            ParseError(
-                detail="Cannot parse response for status code 200 (some problem), "
-                "response will be omitted from generated client",
-                data=parse_error.data,
-            ),
-            ParseError(
-                detail="Cannot parse response for status code 404 (some problem), "
-                "response will be omitted from generated client",
-                data=parse_error.data,
-            ),
-        ]
-
     def test_add_parameters_handles_no_params(self):
         endpoint = self.make_endpoint()
         schemas = Schemas()