✨ Support response envelope.

Author: Raphael Krupinski

ChangeLog.md

@@ -5,10 +5,17 @@ All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html),
 and the format of this file is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+
+## [NEXT]
+### Added
+- Support response envelope objects to allow returning headers together with the body model.
+
+
 ## [0.9.1](https://github.com/python-lapidary/lapidary/releases/tag/v0.9.0) - 2024-05-25
 ### Fixed
 - Moved pytest-asyncio dependency to dev group.
 
+
 ## [0.9.0](https://github.com/python-lapidary/lapidary/releases/tag/v0.9.0) - 2024-05-16
 ### Added
 - Added a default user-agent header.
@@ -23,6 +30,7 @@ and the format of this file is based on [Keep a Changelog](https://keepachangelo
 - Removed support for invalid HTTP status codes patterns, like 20X ([OAS3.1 4.8.16.2](https://spec.openapis.org/oas/v3.1.0#patterned-fields-0)).
 - Removed Absent class #50.
 
+
 ## [0.8.0](https://github.com/python-lapidary/lapidary/releases/tag/v0.8.0) - 2023-01-02
 ### Added
 - Support for arbitrary specification extensions (x- properties).
@@ -32,19 +40,23 @@ and the format of this file is based on [Keep a Changelog](https://keepachangelo
 - Property cross-validation (e.g. only one property of example and examples is allowed).
 - Bearer security scheme.
 
+
 ## [0.7.3](https://github.com/python-lapidary/lapidary/releases/tag/v0.7.3) - 2022-12-15
 ### Fixed
 - None error on missing x-lapidary-responses-global
 - Enum params are rendered as their string representation instead of value.
 
+
 ## [0.7.2](https://github.com/python-lapidary/lapidary/releases/tag/v0.7.2) - 2022-12-15
 ### Fixed
 - platformdirs dependency missing from pyproject.
 
+
 ## [0.7.1](https://github.com/python-lapidary/lapidary/releases/tag/v0.7.1) - 2022-12-15
 ### Fixed
 - Error while constructing a type.
 
+
 ## [0.7.0](https://github.com/python-lapidary/lapidary/releases/tag/v0.7.0) - 2022-12-15
 ### Added
 - Support for api responses.

docs/usage/operation.md

@@ -171,15 +171,15 @@ to JSON using Pydantic's BaseModel.model_dump_json() and included in the body of
 ## Return type
 
 The Responses annotation plays a crucial role in mapping HTTP status codes and Content-Type headers to specific return
-types. This mechanism allows developers to define how responses are parsed and returned with high precision.
+types. This mechanism allows developers to define how responses are parsed and returned.
 
-The return type is specified in two essential places:
+The return type is specified in two places:
 
 1. At the method signature level - The declared return type here should reflect the expected successful response
    structure. It can be a single type or a Union of types, accommodating various potential non-error response bodies.
 
-2. Within the Responses annotation - This details the specific type or types to be used for parsing the response body,
-   contingent upon the response's HTTP status code and content type matching those specified.
+2. Within the `Responses` annotation - This details the specific type or types to be used for parsing the response body,
+   depending on the response's HTTP status code and content type matching those specified.
 
 !!! Note
 
@@ -189,12 +189,7 @@ Example:
 
 ```python
 @get('/cat')
-async def list_cats(
-        self: Self,
-        cat: Annotated[Cat, RequestBody({
-            'application/json': Cat,
-        })],
-) -> Annotated[
+async def list_cats(self: Self) -> Annotated[
     List[Cat],
     Responses({
         '2XX': {
@@ -210,6 +205,38 @@ application/json, the response body will be parsed as a list of Cat objects. Thi
 method's return type is tightly coupled with the anticipated successful response structure, providing clarity and type
 safety for API interactions.
 
+
+### Mapping headers
+
+Lapidary supports an additional response type that envelops the response body and allows declaring response headers.
+
+Example:
+
+```python
+
+class CatsListResponse(ResponseEnvelope):
+   body: Annotated[List[Cat], ResponseBody()]
+   total_count: Annotated[int, ResponseHeader('X-Total-Count')]
+
+class CatClient(ClientBase):
+   @get('/cat')
+   async def list_cats(self: Self) -> Annotated[
+       CatsListResponse,
+       Responses({
+           '2XX': {
+               'application/json': CatsListResponse,
+           },
+       })
+   ]:
+       pass
+
+client = CatClient()
+cats_response = await client.list_cats()
+assert cats_response.body == [Cat(...)]
+assert cats_response.count == 1
+```
+
+
 ### Handling error responses
 
 Lapidary allows you to map specific responses to exceptions. This feature is commonly used for error responses, like

src/lapidary/runtime/__init__.py

@@ -4,6 +4,9 @@
     'NamedAuth',
     'ParamStyle',
     'RequestBody',
+    'ResponseEnvelope',
+    'ResponseHeader',
+    'ResponseBody',
     'Responses',
     'SecurityRequirements',
     'delete',
@@ -22,8 +25,8 @@
 from .client_base import ClientBase
 from .model import ModelBase
 from .model.encode_param import ParamStyle
-from .model.params import RequestBody
-from .model.response_map import Responses
+from .model.params import Body as RequestBody
+from .model.response import Body as ResponseBody, Header as ResponseHeader, ResponseEnvelope, Responses
 from .operation import delete, get, head, patch, post, put, trace
 from .param import Cookie, Header, Path, Query
 from .types_ import NamedAuth, SecurityRequirements

src/lapidary/runtime/model/op.py

@@ -1,12 +1,15 @@
 import dataclasses as dc
 import inspect
+from collections.abc import Sequence
+from typing import Union
 
 import httpx
+import pydantic.fields
 import typing_extensions as typing
 
-from ..response import find_type, parse_model
+from ..response import find_type
 from .params import ParameterAnnotation, RequestPartHandler, find_annotations, process_params
-from .response_map import ResponseMap, Responses
+from .response import DefaultEnvelope, PropertyAnnotation, ResponseEnvelope, ResponseMap, ResponsePartHandler, Responses
 
 if typing.TYPE_CHECKING:
     from .request import RequestBuilder
@@ -43,20 +46,43 @@ def handle_response(self, response: httpx.Response) -> typing.Any:
         if typ is None:
             return None
 
-        obj: typing.Any = parse_model(response, typ)
-
-        if isinstance(obj, Exception):
-            raise obj
+        fields = {}
+        for field_name, field_info in typ.model_fields.items():
+            field_info: pydantic.fields.FieldInfo
+            handlers: Sequence[Union[ResponsePartHandler, type[ResponsePartHandler]]] = [
+                anno
+                for anno in field_info.metadata
+                if isinstance(anno, ResponsePartHandler) or (inspect.isclass(anno) and issubclass(anno, ResponsePartHandler))
+            ]
+            assert len(handlers) == 1
+            handler = handlers[0]
+            if inspect.isclass(handler):
+                handler = handler()
+            if isinstance(handler, PropertyAnnotation):
+                handler.supply_formal(field_name, field_info.annotation)
+            handler.apply(fields, response)
+        obj = typ.parse_obj(fields)
+        # obj: typing.Any = parse_model(response, typ)
+
+        if isinstance(obj, DefaultEnvelope):
+            if isinstance(obj.body, Exception):
+                raise obj.body
+            return obj.body
         else:
             return obj
 
 
 def get_response_map(return_anno: type) -> ResponseMap:
-    annos = find_annotations(return_anno, Responses)
+    annos: typing.Sequence[Responses] = find_annotations(return_anno, Responses)
     if len(annos) != 1:
         raise TypeError('Operation function must have exactly one Responses annotation')
 
-    return annos[0].responses
+    responses = annos[0].responses
+    for media_type_map in responses.values():
+        for media_type, typ in media_type_map.items():
+            if not issubclass(typ, ResponseEnvelope):
+                media_type_map[media_type] = DefaultEnvelope[typ]
+    return responses
 
 
 def get_operation_model(

src/lapidary/runtime/model/params.py

@@ -70,7 +70,7 @@ def __eq__(self, other):
 
 
 @dc.dataclass
-class RequestBody(RequestPartHandler, ParameterAnnotation):
+class Body(RequestPartHandler, ParameterAnnotation):
     name: str = dc.field(init=False)
     content: typing.Mapping[str, type]
     _serializer: pydantic.TypeAdapter = dc.field(init=False)

src/lapidary/runtime/model/response.py

@@ -0,0 +1,74 @@
+import abc
+import dataclasses as dc
+from collections.abc import Callable
+from typing import Any
+
+import httpx
+import pydantic
+import typing_extensions as typing
+
+T = typing.TypeVar('T')
+MimeType = str
+ResponseCode = str
+
+MimeMap = typing.MutableMapping[MimeType, type]
+ResponseMap = typing.Mapping[ResponseCode, MimeMap]
+
+
+@dc.dataclass
+class Responses:
+    responses: ResponseMap
+
+
+class ResponseEnvelopeBuilder:
+    def __init__(self, typ: type[pydantic.BaseModel]) -> None:
+        self._type = typ
+        self.fields = {}
+
+    def build(self) -> Any:
+        return self._type.model_construct(**self.fields)
+
+
+class ResponsePartHandler(abc.ABC):
+    @abc.abstractmethod
+    def apply(self, fields: typing.MutableMapping, response: httpx.Response) -> None:
+        pass
+
+
+class PropertyAnnotation(abc.ABC):
+    _name: str
+    _type: type
+
+    def supply_formal(self, name: str, type_: type) -> None:
+        self._name = name
+        self._type = type_
+
+
+class Body(ResponsePartHandler, PropertyAnnotation):
+    _parse: Callable[[...], Any]
+
+    def supply_formal(self, name: str, type_: type) -> None:
+        super().supply_formal(name, type_)
+        self._parse = pydantic.TypeAdapter(type_).validate_json
+
+    def apply(self, fields: typing.MutableMapping, response: httpx.Response) -> None:
+        fields[self._name] = self._parse(response.text)
+
+
+class Header(ResponsePartHandler, PropertyAnnotation):
+    def __init__(self, header: str) -> None:
+        self._header = header
+
+    def apply(self, fields: typing.MutableMapping, response: httpx.Response) -> None:
+        fields[self._name] = response.headers.get(self._header, None)
+
+
+class ResponseEnvelope(abc.ABC, pydantic.BaseModel):
+    """Marker interface for response envelopes."""
+
+
+BodyT = typing.TypeVar('BodyT')
+
+
+class DefaultEnvelope(ResponseEnvelope, typing.Generic[BodyT]):
+    body: typing.Annotated[BodyT, Body]

src/lapidary/runtime/model/response_map.py

@@ -5,7 +5,7 @@
 from .mime import find_mime
 from .model.op import OperationModel
 from .model.request import RequestBuilder, RequestFactory
-from .model.response_map import ResponseMap
+from .model.response import ResponseMap
 
 
 def get_accept_header(response_map: typing.Optional[ResponseMap]) -> typing.Optional[str]:

src/lapidary/runtime/request.py

@@ -1,29 +1,19 @@
-import inspect
 import logging
 
 import httpx
-import pydantic
 import typing_extensions as typing
 
 from .http_consts import CONTENT_TYPE
 from .mime import find_mime
-from .model.response_map import ResponseMap
+from .model.response import ResponseEnvelope, ResponseMap
 
 logger = logging.getLogger(__name__)
 
 T = typing.TypeVar('T')
 P = typing.TypeVar('P')
 
 
-def parse_model(response: httpx.Response, typ: type[T]) -> T:
-    if inspect.isclass(typ):
-        if issubclass(typ, pydantic.BaseModel):
-            return typing.cast(T, typ.model_validate_json(response.content))
-
-    return pydantic.TypeAdapter(typ).validate_json(response.content)
-
-
-def find_type(response: httpx.Response, response_map: ResponseMap) -> typing.Optional[type]:
+def find_type(response: httpx.Response, response_map: ResponseMap) -> typing.Optional[type[ResponseEnvelope]]:
     status_code = str(response.status_code)
     if CONTENT_TYPE not in response.headers:
         return None