🚧♻️✨ Consolidate request and response header handling. Support response cookies and add basic support for links.

Author: Raphael Krupinski

src/lapidary/runtime/__init__.py

@@ -5,7 +5,6 @@
     'ParamStyle',
     'RequestBody',
     'ResponseEnvelope',
-    'ResponseHeader',
     'ResponseBody',
     'Responses',
     'SecurityRequirements',
@@ -24,9 +23,16 @@
 
 from .client_base import ClientBase
 from .model import ModelBase
+from .model.annotations import (
+    Cookie,
+    Header,
+    Path,
+    Query,
+    RequestBody,
+    ResponseBody,
+    Responses,
+)
 from .model.encode_param import ParamStyle
-from .model.params import Body as RequestBody
-from .model.response import Body as ResponseBody, Header as ResponseHeader, ResponseEnvelope, Responses
+from .model.response import ResponseEnvelope
 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/client_base.py

@@ -8,8 +8,9 @@
 from typing_extensions import Self
 
 from ._httpx import AuthType
-from .model.op import OperationModel, get_operation_model
+from .model.op import OperationModel
 from .model.request import RequestFactory
+from .operation import get_operation_model
 from .request import build_request
 from .types_ import MultiAuth, NamedAuth, SecurityRequirements
 

src/lapidary/runtime/model/annotations.py

@@ -0,0 +1,215 @@
+import abc
+import dataclasses as dc
+import inspect
+from typing import TYPE_CHECKING, Callable, Optional
+
+import httpx
+import pydantic
+import typing_extensions as typing
+
+from ..http_consts import CONTENT_TYPE, MIME_JSON
+from .encode_param import ParamStyle, get_encode_fn
+from .request import RequestHandler
+from .response import ResponseHandler
+
+if TYPE_CHECKING:
+    from .._httpx import ParamValue
+    from .encode_param import Encoder
+    from .request import RequestBuilder
+
+
+class NameTypeAwareAnnotation(abc.ABC):
+    _name: str
+    _type: type
+
+    def supply_formal(self, name: str, typ: type) -> None:
+        self._name = name
+        self._type = typ
+
+
+@dc.dataclass
+class RequestBody(NameTypeAwareAnnotation, RequestHandler):
+    content: typing.Mapping[str, type]
+    _serializer: Callable[[typing.Any], bytes] = dc.field(init=False)
+
+    def supply_formal(self, name: str, typ: type) -> None:
+        super().supply_formal(name, typ)
+        self._serializer = lambda content: pydantic.TypeAdapter(typ).dump_json(
+            content,
+            by_alias=True,
+            exclude_unset=True,
+        )
+
+    def apply_request(self, builder: 'RequestBuilder', value: typing.Any) -> None:
+        content_type = self.find_content_type(value)
+        builder.headers[CONTENT_TYPE] = content_type
+
+        plain_content_type = content_type[: content_type.index(';')] if ';' in content_type else content_type
+        if plain_content_type == MIME_JSON:
+            builder.content = self._serializer(value)
+        else:
+            raise NotImplementedError(content_type)
+
+    def find_content_type(self, value: typing.Any) -> str:
+        for content_type, typ in self.content.items():
+            origin_type = typing.get_origin(typ) or typ
+            if isinstance(value, origin_type):
+                return content_type
+
+        raise ValueError('Could not determine content type')
+
+
+class ResponseBody(NameTypeAwareAnnotation, ResponseHandler):
+    """Annotate response body within an Envelope type."""
+
+    _parse: Callable[[bytes], typing.Any]
+
+    def supply_formal(self, name: str, typ: type) -> None:
+        super().supply_formal(name, typ)
+        self._parse = pydantic.TypeAdapter(typ).validate_json
+
+    def apply_response(self, response: httpx.Response, fields: typing.MutableMapping) -> None:
+        fields[self._name] = self._parse(response.content)
+
+
+class Param(RequestHandler, NameTypeAwareAnnotation, abc.ABC):
+    style: ParamStyle
+    alias: typing.Optional[str]
+    explode: typing.Optional[bool]
+    _encoder: 'typing.Optional[Encoder]'
+
+    def __init__(self, alias: Optional[str], /, *, style: ParamStyle, explode: Optional[bool]) -> None:
+        self.alias = alias
+        self.style = style
+        self.explode = explode
+        self._encoder = None
+
+    def _get_explode(self) -> bool:
+        return self.explode or self.style == ParamStyle.form
+
+    @abc.abstractmethod
+    def _apply_request(self, builder: 'RequestBuilder', value: 'ParamValue'):
+        pass
+
+    def apply_request(self, builder: 'RequestBuilder', value: typing.Any) -> None:
+        if not value:
+            return
+
+        if not self._encoder:
+            self._encoder = get_encode_fn(self._type, self.style, self._get_explode())
+        assert self._encoder
+        value = self._encoder(self._name, value)
+        self._apply_request(builder, value)
+
+    @property
+    def http_name(self) -> str:
+        return self.alias or self._name
+
+    def __eq__(self, other):
+        if not isinstance(other, type(self)):
+            raise NotImplementedError
+        return self.__dict__ == other.__dict__
+
+
+T = typing.TypeVar('T')
+
+
+def find_annotations(
+    user_type: type,
+    annotation_type: type[T],
+) -> typing.Sequence[T]:
+    if user_type is inspect.Signature.empty or '__metadata__' not in dir(user_type):
+        return ()
+    return [anno for anno in user_type.__metadata__ if isinstance(anno, annotation_type)]  # type: ignore[attr-defined]
+
+
+MimeType: typing.TypeAlias = str
+ResponseCode: typing.TypeAlias = str
+MimeMap: typing.TypeAlias = typing.MutableMapping[MimeType, type]
+ResponseMap: typing.TypeAlias = typing.Mapping[ResponseCode, MimeMap]
+
+
+@dc.dataclass
+class Responses:
+    responses: ResponseMap
+
+
+class Header(Param, ResponseHandler):
+    def __init__(
+        self,
+        alias: Optional[str] = None,
+        /,
+        *,
+        style: ParamStyle = ParamStyle.simple,
+        explode: Optional[bool] = None,
+    ) -> None:
+        super().__init__(alias, style=style, explode=explode)
+
+    def _apply_request(self, builder: 'RequestBuilder', value: 'ParamValue') -> None:
+        builder.headers[self.http_name] = str(value)
+
+    def apply_response(self, response: 'httpx.Response', fields: typing.MutableMapping[str, typing.Any]) -> None:
+        # TODO decode
+        if self.http_name in response.headers:
+            fields[self._name] = response.headers[self.http_name]
+
+
+class Cookie(Param, ResponseHandler):
+    def __init__(
+        self,
+        alias: Optional[str] = None,
+        /,
+        *,
+        style: ParamStyle = ParamStyle.form,
+        explode: Optional[bool] = None,
+    ) -> None:
+        super().__init__(alias, style=style, explode=explode)
+
+    def _apply_request(self, builder: 'RequestBuilder', value: typing.Any) -> None:
+        builder.cookies[self.http_name] = value
+
+    def apply_response(self, response: 'httpx.Response', fields: typing.MutableMapping[str, typing.Any]) -> None:
+        # TODO handle decoding
+        if self.http_name in response.headers:
+            fields[self._name] = response.cookies[self.http_name]
+
+
+class Path(Param):
+    def __init__(
+        self,
+        alias: Optional[str] = None,
+        /,
+        *,
+        style: ParamStyle = ParamStyle.simple,
+        explode: Optional[bool] = None,
+    ) -> None:
+        super().__init__(alias, style=style, explode=explode)
+
+    def _apply_request(self, builder: 'RequestBuilder', value: typing.Any) -> None:
+        builder.path_params[self.http_name] = value
+
+
+class Query(Param):
+    def __init__(
+        self,
+        alias: Optional[str] = None,
+        /,
+        *,
+        style: ParamStyle = ParamStyle.form,
+        explode: Optional[bool] = None,
+    ) -> None:
+        super().__init__(alias, style=style, explode=explode)
+
+    def _apply_request(self, builder: 'RequestBuilder', value: typing.Any) -> None:
+        builder.query_params.append((self.http_name, value))
+
+
+class Link(NameTypeAwareAnnotation, ResponseHandler):
+    def __init__(self, name: str) -> None:
+        self._link_name = name
+
+    def apply_response(self, response: httpx.Response, fields: typing.MutableMapping[str, typing.Any]) -> None:
+        if 'Link' in response.headers:
+            links = response.links
+            if self._link_name in links:
+                fields[self._name] = links[self._link_name]

src/lapidary/runtime/model/op.py

@@ -1,13 +1,13 @@
 import dataclasses as dc
-import inspect
 from collections.abc import Sequence
 
 import httpx
 import typing_extensions as typing
 
 from ..response import find_type
-from .params import ParameterAnnotation, RequestPartHandler, find_annotations, process_params
-from .response import DefaultEnvelope, PropertyAnnotation, ResponseEnvelope, ResponseMap, ResponsePartHandler, Responses
+from .annotations import NameTypeAwareAnnotation, ResponseBody, ResponseMap
+from .request import RequestHandler
+from .response import ResponseEnvelope, ResponseHandler
 
 if typing.TYPE_CHECKING:
     from .request import RequestBuilder
@@ -17,7 +17,7 @@
 class OperationModel:
     method: str
     path: str
-    params: typing.Mapping[str, ParameterAnnotation]
+    params: typing.Mapping[str, NameTypeAwareAnnotation]
     response_map: ResponseMap
 
     def process_params(
@@ -27,8 +27,8 @@ def process_params(
     ) -> None:
         for param_name, value in actual_params.items():
             param_handler = self.params[param_name]
-            if isinstance(param_handler, RequestPartHandler):
-                param_handler.apply(request, actual_params[param_name])
+            if isinstance(param_handler, RequestHandler):
+                param_handler.apply_request(request, actual_params[param_name])
             else:
                 raise TypeError(param_name, type(value))
 
@@ -46,14 +46,14 @@ def handle_response(self, response: httpx.Response) -> typing.Any:
 
         fields: typing.MutableMapping[str, typing.Any] = {}
         for field_name, field_info in typ.model_fields.items():
-            handlers: Sequence[ResponsePartHandler] = [anno for anno in field_info.metadata if isinstance(anno, ResponsePartHandler)]
+            handlers: Sequence[ResponseHandler] = [anno for anno in field_info.metadata if isinstance(anno, ResponseHandler)]
             assert len(handlers) == 1
             handler = handlers[0]
-            if isinstance(handler, PropertyAnnotation):
+            if isinstance(handler, NameTypeAwareAnnotation):
                 field_type = field_info.annotation
                 assert field_type
                 handler.supply_formal(field_name, field_type)
-            handler.apply(fields, response)
+            handler.apply_response(response, fields)
         obj = typ.parse_obj(fields)
         # obj: typing.Any = parse_model(response, typ)
 
@@ -65,28 +65,8 @@ def handle_response(self, response: httpx.Response) -> typing.Any:
             return obj
 
 
-def get_response_map(return_anno: type) -> ResponseMap:
-    annos: typing.Sequence[Responses] = find_annotations(return_anno, Responses)
-    if len(annos) != 1:
-        raise TypeError('Operation function must have exactly one Responses annotation')
+BodyT = typing.TypeVar('BodyT')
 
-    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]  # type: ignore[valid-type]
-    return responses
 
-
-def get_operation_model(
-    method: str,
-    path: str,
-    fn: typing.Callable,
-) -> OperationModel:
-    sig = inspect.signature(fn)
-    return OperationModel(
-        method=method,
-        path=path,
-        params=process_params(sig),
-        response_map=get_response_map(sig.return_annotation),
-    )
+class DefaultEnvelope(ResponseEnvelope, typing.Generic[BodyT]):
+    body: typing.Annotated[BodyT, ResponseBody()]