fix: Send a POST request as application/json when body is a Python dictionary (#160)

Author: pmateusz

src/meatie/internal/template/request.py

@@ -123,7 +123,11 @@ def build_request(self, *args: Any, **kwargs: Any) -> Request:
 
             if param.kind == Kind.BODY:
                 if param.formatter is not None:
-                    body_data = param.formatter(value)
+                    raw_value = param.formatter(value)
+                    if isinstance(raw_value, (str, bytes)):
+                        body_data = raw_value
+                    else:
+                        body_json = raw_value
                 else:
                     body_json = self.request_encoder.to_content(value)
                 continue

tests/client/aiohttp/adapter/test_pydantic_v2.py

@@ -1,53 +1,54 @@
 #  Copyright 2024 The Meatie Authors. All rights reserved.
 #  Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.
-import json
 from http import HTTPStatus
-from http.server import BaseHTTPRequestHandler
-from typing import Annotated, Any, Optional, Union
+from typing import Annotated, Union
 
 import pytest
 from aiohttp import ClientSession
 from http_test import Handler, HTTPTestServer
+from http_test.handlers import echo_json_handler
 from typing_extensions import Literal
 
-from meatie import AsyncResponse, api_ref, endpoint
+from meatie import api_ref, endpoint
 from meatie_aiohttp import Client
 
 pydantic = pytest.importorskip("pydantic", minversion="2.0.0")
 
 
-@pytest.mark.asyncio()
-async def test_post_request_body_with_fmt(http_server: HTTPTestServer) -> None:
-    # GIVEN
-    class Request(pydantic.BaseModel):
-        data: Optional[dict[str, Any]]
+class Todo(pydantic.BaseModel):
+    user_id: int = pydantic.Field(alias="userId")
+    id: int
+    title: str
+    completed: bool
 
-    def handler(request: BaseHTTPRequestHandler) -> None:
-        content_length = request.headers.get("Content-Length", "0")
-        raw_body = request.rfile.read(int(content_length))
-        body = json.loads(raw_body.decode("utf-8"))
 
-        if body.get("data") is not None:
-            request.send_response(HTTPStatus.BAD_REQUEST)
-        else:
-            request.send_response(HTTPStatus.OK)
-        request.end_headers()
+class Params:
+    @staticmethod
+    def todo(value: Todo) -> dict:
+        return value.model_dump(by_alias=True)
 
-    http_server.handler = handler
 
-    def dump_body(model: pydantic.BaseModel) -> str:
-        return model.model_dump_json(by_alias=True, exclude_none=True)
+class JsonPlaceholderClient(Client):
+    def __init__(self, base_url: str) -> None:
+        super().__init__(ClientSession(base_url=base_url))
 
-    class TestClient(Client):
-        @endpoint("/")
-        async def post_request(self, body: Annotated[Request, api_ref(fmt=dump_body)]) -> AsyncResponse: ...
+    @endpoint("/todos")
+    async def post_todo_as_dict(
+        self, todo: Annotated[Todo, api_ref("body", fmt=lambda data: data.model_dump(by_alias=True))]
+    ) -> Todo: ...
+
+
+@pytest.mark.asyncio()
+async def test_post_request_body_with_fmt_as_dict(http_server: HTTPTestServer) -> None:
+    # GIVEN
+    http_server.handler = echo_json_handler
 
     # WHEN
-    async with TestClient(ClientSession(http_server.base_url)) as client:
-        response = await client.post_request(Request(data=None))
+    async with JsonPlaceholderClient(http_server.base_url) as client:
+        todo = await client.post_todo_as_dict(Todo(userId=123, id=456, title="abc", completed=True))
 
     # THEN
-    assert response.status == HTTPStatus.OK
+    assert todo.user_id == 123
 
 
 class TodoV1(pydantic.BaseModel):

tests/client/httpx/adapter/test_pydantic_v2.py

@@ -0,0 +1,48 @@
+#  Copyright 2024 The Meatie Authors. All rights reserved.
+#  Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.
+from typing import Annotated
+
+import pytest
+from http_test import HTTPTestServer
+from http_test.handlers import echo_json_handler
+from httpx import Client as HttpxClient
+
+from meatie import api_ref, endpoint
+from meatie_httpx import Client
+
+pydantic = pytest.importorskip("pydantic", minversion="2.0.0")
+
+
+class Todo(pydantic.BaseModel):
+    user_id: int = pydantic.Field(alias="userId")
+    id: int
+    title: str
+    completed: bool
+
+
+class Params:
+    @staticmethod
+    def todo(value: Todo) -> dict:
+        return value.model_dump(by_alias=True)
+
+
+class JsonPlaceholderClient(Client):
+    def __init__(self, base_url: str) -> None:
+        super().__init__(HttpxClient(base_url=base_url))
+
+    @endpoint("/todos")
+    def post_todo_as_dict(
+        self, todo: Annotated[Todo, api_ref("body", fmt=lambda data: data.model_dump(by_alias=True))]
+    ) -> Todo: ...
+
+
+def test_post_request_body_with_fmt_as_dict(http_server: HTTPTestServer) -> None:
+    # GIVEN
+    http_server.handler = echo_json_handler
+
+    # WHEN
+    with JsonPlaceholderClient(http_server.base_url) as client:
+        todo = client.post_todo_as_dict(Todo(userId=123, id=456, title="abc", completed=True))
+
+    # THEN
+    assert todo.user_id == 123

tests/client/requests/adapter/test_pydantic_v2.py

@@ -0,0 +1,48 @@
+#  Copyright 2024 The Meatie Authors. All rights reserved.
+#  Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.
+from typing import Annotated
+
+import pytest
+from http_test import HTTPTestServer
+from http_test.handlers import echo_json_handler
+from requests import Session
+
+from meatie import api_ref, endpoint
+from meatie_requests import Client
+
+pydantic = pytest.importorskip("pydantic", minversion="2.0.0")
+
+
+class Todo(pydantic.BaseModel):
+    user_id: int = pydantic.Field(alias="userId")
+    id: int
+    title: str
+    completed: bool
+
+
+class Params:
+    @staticmethod
+    def todo(value: Todo) -> dict:
+        return value.model_dump(by_alias=True)
+
+
+class JsonPlaceholderClient(Client):
+    def __init__(self, base_url: str) -> None:
+        super().__init__(Session(), prefix=base_url)
+
+    @endpoint("/todos")
+    def post_todo_as_dict(
+        self, todo: Annotated[Todo, api_ref("body", fmt=lambda data: data.model_dump(by_alias=True))]
+    ) -> Todo: ...
+
+
+def test_post_request_body_with_fmt_as_dict(http_server: HTTPTestServer) -> None:
+    # GIVEN
+    http_server.handler = echo_json_handler
+
+    # WHEN
+    with JsonPlaceholderClient(http_server.base_url) as client:
+        todo = client.post_todo_as_dict(Todo(userId=123, id=456, title="abc", completed=True))
+
+    # THEN
+    assert todo.user_id == 123

tests/shared/http_test/handlers.py

@@ -5,6 +5,7 @@
 from decimal import Decimal
 from http import HTTPStatus
 from http.server import BaseHTTPRequestHandler, SimpleHTTPRequestHandler
+from json import JSONDecodeError
 from typing import Any, Callable, Optional
 from urllib.parse import parse_qs, urlparse
 
@@ -91,6 +92,24 @@ def echo_handler(handler: Handler) -> None:
     handler.send_bytes(HTTPStatus.OK, content_type, raw_body_opt)
 
 
+def echo_json_handler(handler: Handler) -> None:
+    content_type = handler.headers.get("Content-Type")
+    raw_body = handler.safe_text()
+    if content_type != "application/json" or raw_body is None:
+        handler.send_response(HTTPStatus.BAD_REQUEST)
+        handler.end_headers()
+        return
+
+    try:
+        body = json.loads(raw_body)
+    except JSONDecodeError:
+        handler.send_response(HTTPStatus.BAD_REQUEST)
+        handler.end_headers()
+        return
+
+    handler.send_json(HTTPStatus.OK, body)
+
+
 def diagnostic_handler(handler: Handler) -> None:
     body_opt = handler.safe_text()
     if body_opt is None: