🐛 Fix serializing collection request body.

Author: rafalkrupinski

ChangeLog.md

@@ -5,12 +5,16 @@ 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/).
 
+
 ## [Unreleased]
 
 ### Added
-
 - Accept session_factory in `ClientBase.__init__`.
-- Python 3.13 is not supported (after upgrading mimeparse to 2.0.0rc1).
+
+### Fixed
+- Handling collections in request bodies.
+- Dropped dependency on std package cgi, which allows running under python 3.13 .
+
 
 ## [0.11.0] - 2024-08-13
 

docs/adr/2024-10-17 serializing body.md

@@ -0,0 +1,30 @@
+# Serializing request body
+
+## The Current solution (0.11.0)
+
+Currently lapidary checks type of value of the body parameter and tries to match against the type map passed to the `Body` annotation.
+
+Problem starts when body is a collection: `type()` returns the type of only the collection (e.g. `list`), so the type matching fails.
+
+## Possible alternatives
+
+1. Check the type of the first item, but there's never a guarantee that the passed collection is homogenic.
+Both JSON Schema and python typing support heterogenic collections.
+
+2. Check type of all items is out of the question for performance reasons, and pydantic does it anyway during serialization.
+
+3. Try to serialize the value with a TypeAdapter for each type in the type map. The first successful attempt also determines the body content type.
+4. Either accept extra parameter `body_type: type` or accept body as tuple with the type explicitly declared: `body: T | Union[T, type]`.
+
+The last two solutions seem feasible.
+Trying every type would incur a performance hit for unions of complex types, but
+- it would handle simpler cases well,
+- keep lapidary compatible with lapidary-render,
+
+Lapidary could still accept optional type parameter but use the other method as a fallback for when user doesn't pass the type.
+
+# Accepted solution
+
+Lapidary will not check the type of body value, instead it will try serializing it with every type mentioned in the type map in `Body` annotation.
+
+Lapidary should implement an optional explicit body type parameter in a future version, either as a separate parameter, or as a tuple together with the body value.

src/lapidary/runtime/model/request.py

@@ -12,7 +12,6 @@
 from ..annotations import Body, Cookie, Header, Metadata, Param, Path, Query, WebArg
 from ..http_consts import ACCEPT, CONTENT_TYPE, MIME_JSON
 from ..metattype import is_array_like, make_not_optional
-from ..mime import find_mime
 from ..types_ import Dumper, MimeType, RequestFactory, SecurityRequirements
 from .annotations import (
     find_annotation,
@@ -23,6 +22,9 @@
 if typing.TYPE_CHECKING:
     from ..client_base import ClientBase
     from ..operation import Operation
+import logging
+
+logger = logging.getLogger(__name__)
 
 
 @dc.dataclass
@@ -174,37 +176,41 @@ def update_builder(self, builder: RequestBuilder, free_params: Mapping[str, typi
 
 @dc.dataclass
 class BodyContributor:
-    dumpers: Mapping[type, Mapping[MimeType, Dumper]]
+    serializers: list[tuple[pydantic.TypeAdapter, str]]
 
     def update_builder(self, builder: 'RequestBuilder', value: typing.Any, media_type: MimeType = MIME_JSON) -> None:
-        matched_media_type, dump = self.find_dumper(type(value), media_type)
+        matched_media_type, content = self._dump(value, media_type)
         builder.headers[CONTENT_TYPE] = matched_media_type
-        builder.content = dump(value)
+        builder.content = content
 
-    def find_dumper(self, typ: type, media_type: MimeType) -> tuple[MimeType, Dumper]:
-        for base in typ.__mro__[:-1]:
+    def _dump(self, value: typing.Any, media_type: MimeType = MIME_JSON):
+        for type_adapter, media_type_ in self.serializers:
+            if not BodyContributor._media_matches(media_type_, media_type):
+                logger.debug('Ignoring unsupported media_type: %s', media_type)
+                continue
             try:
-                media_dumpers = self.dumpers[base]
-                matched_media_type = find_mime(media_dumpers.keys(), media_type)
-                if not matched_media_type:
-                    raise ValueError('Unsupported media_type', media_type)
-                return matched_media_type, media_dumpers[matched_media_type]
-            except KeyError:
+                raw = type_adapter.dump_json(value, exclude_unset=True, by_alias=True)
+                return media_type_, raw
+            except pydantic.ValidationError:
                 continue
-        raise TypeError(f'Unsupported type: {typ.__name__}')
+        else:
+            raise ValueError('Unsupported value')
 
     @classmethod
     def for_parameter(cls, annotation: type) -> typing.Self:
-        dumpers: dict[type, dict[MimeType, Dumper]] = {}
         body: Body
         _, body = find_annotation(annotation, Body)
-        for media_type, typ in body.content.items():
-            m_type, m_subtype, _ = mimeparse.parse_media_range(media_type)
-            if f'{m_type}/{m_subtype}' != MIME_JSON:
-                raise TypeError(f'Unsupported media type: {media_type}')
-            type_dumpers = dumpers.setdefault(typ, {})
-            type_dumpers[media_type] = mk_pydantic_dumper(typ)
-        return cls(dumpers=dumpers)
+        serializers = [
+            (pydantic.TypeAdapter(python_type), media_type)
+            for media_type, python_type in body.content.items()
+            if BodyContributor._media_matches(media_type)
+        ]
+        return cls(serializers)
+
+    @staticmethod
+    def _media_matches(media_type: str, match: str = MIME_JSON) -> bool:
+        m_type, m_subtype, _ = mimeparse.parse_media_range(media_type)
+        return f'{m_type}/{m_subtype}' == match
 
 
 @dc.dataclass

tests/test_body_serialization.py

@@ -0,0 +1,128 @@
+from typing import Annotated, Generic, Optional, Union
+
+from client import ClientTestBase
+from httpx import AsyncClient
+from typing_extensions import Self, TypeVar
+
+from lapidary.runtime import Body, ModelBase, Responses, get
+from lapidary.runtime.model.op import process_operation_method
+
+
+class BodyModel(ModelBase):
+    a: Optional[str]
+
+
+def test_serialize_str():
+    class Client(ClientTestBase):
+        def op(self: Self, body: Annotated[str, Body({'application/json': str})]) -> Annotated[None, Responses({})]:
+            pass
+
+    client = ClientTestBase(AsyncClient())
+
+    adapter, response = process_operation_method(Client.op, get('/path'))
+    request, auth = adapter.build_request(client, dict(body='a'))
+
+    assert request.content == b'"a"'
+
+
+def test_serialize_obj():
+    class Client(ClientTestBase):
+        def op(self: Self, body: Annotated[BodyModel, Body({'application/json': BodyModel})]) -> Annotated[None, Responses({})]:
+            pass
+
+    client = ClientTestBase(AsyncClient())
+
+    adapter, response = process_operation_method(Client.op, get('/path'))
+    request, auth = adapter.build_request(client, dict(body=BodyModel(a='a')))
+
+    assert request.content == b'{"a":"a"}'
+
+
+def test_serialize_list():
+    class Client(ClientTestBase):
+        def op(self: Self, body: Annotated[list[BodyModel], Body({'application/json': list[BodyModel]})]) -> Annotated[None, Responses({})]:
+            pass
+
+    client = ClientTestBase(AsyncClient())
+
+    adapter, response = process_operation_method(Client.op, get('/path'))
+    request, auth = adapter.build_request(client, dict(body=[BodyModel(a='a')]))
+
+    assert request.content == b'[{"a":"a"}]'
+
+
+T = TypeVar('T')
+
+
+class GenericBodyModel(ModelBase, Generic[T]):
+    a: T
+
+
+def test_serialize_generic_str():
+    class Client(ClientTestBase):
+        def op(
+            self: Self, body: Annotated[list[GenericBodyModel[str]], Body({'application/json': list[GenericBodyModel[str]]})]
+        ) -> Annotated[None, Responses({})]:
+            pass
+
+    client = ClientTestBase(AsyncClient())
+
+    adapter, response = process_operation_method(Client.op, get('/path'))
+    request, auth = adapter.build_request(client, dict(body=[GenericBodyModel(a='a')]))
+
+    assert request.content == b'[{"a":"a"}]'
+
+
+def test_serialize_generic_int():
+    class Client(ClientTestBase):
+        def op(
+            self: Self, body: Annotated[list[GenericBodyModel[int]], Body({'application/json': list[GenericBodyModel[int]]})]
+        ) -> Annotated[None, Responses({})]:
+            pass
+
+    client = ClientTestBase(AsyncClient())
+
+    adapter, response = process_operation_method(Client.op, get('/path'))
+    request, auth = adapter.build_request(client, dict(body=[GenericBodyModel(a=1)]))
+
+    assert request.content == b'[{"a":1}]'
+
+
+def test_serialize_generic_obj():
+    class Client(ClientTestBase):
+        def op(
+            self: Self, body: Annotated[list[GenericBodyModel[BodyModel]], Body({'application/json': list[GenericBodyModel[BodyModel]]})]
+        ) -> Annotated[None, Responses({})]:
+            pass
+
+    client = ClientTestBase(AsyncClient())
+
+    adapter, response = process_operation_method(Client.op, get('/path'))
+    request, auth = adapter.build_request(client, dict(body=[GenericBodyModel(a=BodyModel(a='a'))]))
+
+    assert request.content == b'[{"a":{"a":"a"}}]'
+
+
+def test_serialize_generic_union():
+    class Client(ClientTestBase):
+        def op(
+            self: Self,
+            body: Annotated[
+                Union[list[GenericBodyModel[BodyModel]], GenericBodyModel[BodyModel], BodyModel],
+                Body({'application/json': Union[list[GenericBodyModel[BodyModel]], GenericBodyModel[BodyModel], BodyModel]}),
+            ],
+        ) -> Annotated[None, Responses({})]:
+            pass
+
+    client = ClientTestBase(AsyncClient())
+
+    adapter, _ = process_operation_method(Client.op, get('/path'))
+
+    request, _ = adapter.build_request(client, dict(body=[GenericBodyModel(a=BodyModel(a='a'))]))
+    assert request.content == b'[{"a":{"a":"a"}}]'
+
+    request, _ = adapter.build_request(client, dict(body=GenericBodyModel(a=BodyModel(a='a'))))
+    assert request.content == b'{"a":{"a":"a"}}'
+
+    request, _ = adapter.build_request(client, dict(body=BodyModel(a='a')))
+    assert request.content == b'{"a":"a"}'