use asgi_tools for HTTP responses

Author: Archmonger

pyproject.toml

@@ -39,6 +39,7 @@ dependencies = [
   "lxml >=4",
   "servestatic >=3.0.0",
   "orjson >=3",
+  "asgi-tools",
 ]
 dynamic = ["version"]
 urls.Changelog = "https://reactpy.dev/docs/about/changelog.html"

src/reactpy/asgi/standalone.py

@@ -8,17 +8,13 @@
 from logging import getLogger
 from typing import Callable, Literal, cast, overload
 
+from asgi_tools import ResponseHTML
 from asgiref import typing as asgi_types
 from typing_extensions import Unpack
 
 from reactpy import html
 from reactpy.asgi.middleware import ReactPyMiddleware
-from reactpy.asgi.utils import (
-    dict_to_byte_list,
-    http_response,
-    import_dotted_path,
-    vdom_head_to_html,
-)
+from reactpy.asgi.utils import import_dotted_path, vdom_head_to_html
 from reactpy.types import (
     AsgiApp,
     AsgiHttpApp,
@@ -40,7 +36,7 @@ def __init__(
         self,
         root_component: RootComponentConstructor,
         *,
-        http_headers: dict[str, str | int] | None = None,
+        http_headers: dict[str, str] | None = None,
         html_head: VdomDict | None = None,
         html_lang: str = "en",
         **settings: Unpack[ReactPyConfig],
@@ -182,44 +178,33 @@ async def __call__(
 
         # Response headers for `index.html` responses
         request_headers = dict(scope["headers"])
-        response_headers: dict[str, str | int] = {
+        response_headers: dict[str, str] = {
             "etag": self._etag,
             "last-modified": self._last_modified,
             "access-control-allow-origin": "*",
             "cache-control": "max-age=60, public",
-            "content-length": len(self._cached_index_html),
+            "content-length": str(len(self._cached_index_html)),
             "content-type": "text/html; charset=utf-8",
             **self.parent.extra_headers,
         }
 
         # Browser is asking for the headers
         if scope["method"] == "HEAD":
-            return await http_response(
-                send=send,
-                method=scope["method"],
-                headers=dict_to_byte_list(response_headers),
-            )
+            response = ResponseHTML("", headers=response_headers)
+            return await response(scope, receive, send)  # type: ignore
 
         # Browser already has the content cached
         if (
             request_headers.get(b"if-none-match") == self._etag.encode()
             or request_headers.get(b"if-modified-since") == self._last_modified.encode()
         ):
             response_headers.pop("content-length")
-            return await http_response(
-                send=send,
-                method=scope["method"],
-                code=304,
-                headers=dict_to_byte_list(response_headers),
-            )
+            response = ResponseHTML("", headers=response_headers, status_code=304)
+            return await response(scope, receive, send)  # type: ignore
 
         # Send the index.html
-        await http_response(
-            send=send,
-            method=scope["method"],
-            message=self._cached_index_html,
-            headers=dict_to_byte_list(response_headers),
-        )
+        response = ResponseHTML(self._cached_index_html, headers=response_headers)
+        await response(scope, receive, send)  # type: ignore
 
     def process_index_html(self) -> None:
         """Process the index.html and store the results in memory."""

src/reactpy/asgi/utils.py

@@ -5,8 +5,6 @@
 from importlib import import_module
 from typing import Any
 
-from asgiref import typing as asgi_types
-
 from reactpy._option import Option
 from reactpy.types import ReactPyConfig, VdomDict
 from reactpy.utils import vdom_to_html
@@ -55,18 +53,6 @@ def check_path(url_path: str) -> str:  # pragma: no cover
     return ""
 
 
-def dict_to_byte_list(
-    data: dict[str, str | int],
-) -> list[tuple[bytes, bytes]]:
-    """Convert a dictionary to a list of byte tuples."""
-    result: list[tuple[bytes, bytes]] = []
-    for key, value in data.items():
-        new_key = key.encode()
-        new_value = value.encode() if isinstance(value, str) else str(value).encode()
-        result.append((new_key, new_value))
-    return result
-
-
 def vdom_head_to_html(head: VdomDict) -> str:
     if isinstance(head, dict) and head.get("tagName") == "head":
         return vdom_to_html(head)
@@ -76,35 +62,6 @@ def vdom_head_to_html(head: VdomDict) -> str:
     )
 
 
-async def http_response(
-    *,
-    send: asgi_types.ASGISendCallable,
-    method: str,
-    code: int = 200,
-    message: str = "",
-    headers: Iterable[tuple[bytes, bytes]] = (),
-) -> None:
-    """Sends a HTTP response using the ASGI `send` API."""
-    start_msg: asgi_types.HTTPResponseStartEvent = {
-        "type": "http.response.start",
-        "status": code,
-        "headers": [*headers],
-        "trailers": False,
-    }
-    body_msg: asgi_types.HTTPResponseBodyEvent = {
-        "type": "http.response.body",
-        "body": b"",
-        "more_body": False,
-    }
-
-    # Add the content type and body to everything other than a HEAD request
-    if method != "HEAD":
-        body_msg["body"] = message.encode()
-
-    await send(start_msg)
-    await send(body_msg)
-
-
 def process_settings(settings: ReactPyConfig) -> None:
     """Process the settings and return the final configuration."""
     from reactpy import config