add method to wrap asgi callbacks in Request

pyproject.toml

@@ -20,6 +20,7 @@ classifiers = [
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
     "Topic :: Internet :: WWW/HTTP",
     "Topic :: Internet :: WWW/HTTP :: Dynamic Content",
     "Topic :: Software Development :: Libraries :: Application Frameworks",
@@ -34,25 +35,28 @@ python = "^3.11"
 python-multipart = "*"
 aiofiles = "^23.2"
 
-[tool.poetry.dev-dependencies]
-uvicorn = { version = "^0.24", extras = ["standard"] }
-pylint = "^3.0"
-flake8 = "^6.1"
-mypy = "^1.5"
+[tool.poetry.group.dev]
+optional = true
+
+[tool.poetry.group.dev.dependencies]
+uvicorn = { version = "^0.29", extras = ["standard"] }
+pylint = "^3.1"
+flake8 = "^7.0"
+mypy = "^1.9"
 isort = "^5.13"
-black = "^23.12"
-asgiref = "^3.7"
-ruff = "^0.1"
+black = "^24.3"
+asgiref = "^3.8"
+ruff = "^0.3"
 
 [tool.poetry.group.test]
 optional = true
 
 [tool.poetry.group.test.dependencies]
-pytest = "^7.4"
+pytest = "^8.1"
 pytest-asyncio = "^0.23"
-pytest-cov = "^4.1"
-coverage = { version = "^7.3", extras = ["toml"] }
-httpx = "^0.25"
+pytest-cov = "^5.0"
+coverage = { version = "^7.4", extras = ["toml"] }
+httpx = "^0.27"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"

src/asgikit/requests.py

@@ -1,7 +1,8 @@
 import asyncio
 import json
 import re
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterable, Awaitable, Callable
+from functools import partial
 from http import HTTPMethod
 from http.cookies import SimpleCookie
 from typing import Any
@@ -110,6 +111,7 @@ def scheme(self):
     @property
     def method(self) -> HTTPMethod | None:
         """Return None when request is websocket"""
+
         if method := self.asgi.scope.get("method"):
             return HTTPMethod(method)
 
@@ -170,6 +172,18 @@ def charset(self) -> str:
     def accept(self):
         return self.headers.get_all("accept")
 
+    def wrap_asgi(
+        self,
+        *,
+        receive: Callable[[AsgiReceive], Awaitable] = None,
+        send: Callable[[AsgiSend, dict], Awaitable] = None,
+    ):
+        new_receive = (
+            partial(receive, self.asgi.receive) if receive else self.asgi.receive
+        )
+        new_send = partial(send, self.asgi.send) if send else self.asgi.send
+        self.asgi = AsgiProtocol(self.asgi.scope, new_receive, new_send)
+
     def __getitem__(self, item):
         return self.attributes[item]
 

src/asgikit/responses.py

@@ -231,9 +231,9 @@ async def end(self):
 
 
 async def respond_text(
-    response: Response, content: str, *, status: HTTPStatus = HTTPStatus.OK
+    response: Response, content: str | bytes, *, status: HTTPStatus = HTTPStatus.OK
 ):
-    data = content.encode(response.encoding)
+    data = content.encode(response.encoding) if isinstance(content, str) else content
     if not response.content_type:
         response.content_type = "text/plain"
 

tests/test_integration.py

@@ -1,4 +1,4 @@
-from httpx import AsyncClient
+from httpx import ASGITransport, AsyncClient
 
 from asgikit.requests import Request
 from asgikit.responses import respond_text
@@ -9,6 +9,6 @@ async def app(scope, receive, send):
         request = Request(scope, receive, send)
         await respond_text(request.response, "Ok")
 
-    client = AsyncClient(app=app)
+    client = AsyncClient(transport=ASGITransport(app))
     response = await client.get("http://localhost:8000/")
     assert response.text == "Ok"

tests/test_requests.py

@@ -213,3 +213,33 @@ def test_request_edit_attributes():
 
     assert "str" not in request
     assert "int" not in request
+
+
+async def test_request_wrap_asgi():
+    async def receive() -> dict:
+        return {}
+
+    async def send(_: dict):
+        pass
+
+    send_event = {}
+
+    request = Request(copy.copy(SCOPE), receive, send)
+
+    assert await request.asgi.receive() == {}
+    await request.asgi.send(send_event)
+    assert send_event == {}
+
+    async def new_receive(orig_receive) -> dict:
+        event = await orig_receive()
+        return event | {"wrapped": True}
+
+    async def new_send(orig_send, event: dict):
+        event["wrapped"] = True
+        await orig_send(event)
+
+    request.wrap_asgi(receive=new_receive, send=new_send)
+
+    assert await request.asgi.receive() == {"wrapped": True}
+    await request.asgi.send(send_event)
+    assert send_event == {"wrapped": True}