Add an options argument to the ASGI and WSGI adapters

workers/src/workers/_adapter.py

@@ -3,6 +3,7 @@
 from collections.abc import Iterable
 from typing import Any
 
+from .gateway import BaseOptions
 from .request import Request
 
 # 101/103/204/205/304 must not carry a body per the Fetch spec.
@@ -15,8 +16,16 @@ def request_headers(req: Any) -> Iterable[tuple[str, str]]:
     return req.headers.items() if isinstance(req, Request) else req.headers
 
 
-def body_encoding(headers: Iterable[tuple[str, str]]) -> dict[str, str]:
-    """The `Response` init entry for a body the app has already encoded."""
+def body_encoding(
+    headers: Iterable[tuple[str, str]], options: BaseOptions | None
+) -> dict[str, Any]:
+    """The `Response` init entry for the body's `Content-Encoding`.
+
+    Explicit `options` behave as in the stock SDK. Without them, a body sent
+    with a `Content-Encoding` header counts as already encoded.
+    """
+    if options is not None:
+        return options.response_init()
     if any(name.lower() == "content-encoding" for name, _ in headers):
         # The app encoded the body already, as GZipMiddleware does. Without
         # this the runtime encodes it a second time.

workers/src/workers/asgi.py

@@ -10,6 +10,7 @@
 from asyncio import Event, Future, Queue, create_task, ensure_future
 from collections.abc import Awaitable
 from contextlib import contextmanager, suppress
+from dataclasses import dataclass
 from typing import Any
 from urllib.parse import unquote
 
@@ -19,6 +20,7 @@
 from ._js import to_js_headers as _to_js_headers
 from .context import ExecutionContext, js_promise
 from .entrypoints import WorkerEntrypoint
+from .gateway import BaseOptions
 from .request import Request
 from .types import Context
 
@@ -27,6 +29,11 @@
 background_tasks = set()
 
 
+@dataclass(frozen=True, kw_only=True)
+class Options(BaseOptions):
+    """Options for the ASGI adapter."""
+
+
 def run_in_background(coro: Awaitable[Any]) -> Future:
     fut = ensure_future(coro)
     background_tasks.add(fut)
@@ -194,8 +201,9 @@ async def start_application(app):
 class _HttpCycle:
     """One HTTP request through an ASGI app, and the `js.Response` it sends."""
 
-    def __init__(self, app: Any) -> None:
+    def __init__(self, app: Any, options: Options | None) -> None:
         self.app = app
+        self.options = options
         self.status: int | None = None
         self.headers: list[tuple[str, str]] | None = None
         self.result: Future[js.Response] = Future()
@@ -273,7 +281,7 @@ def _response_init(self) -> dict[str, Any]:
         return {
             "headers": _to_js_headers(self.headers),
             "status": self.status,
-            **body_encoding(self.headers or []),
+            **body_encoding(self.headers or [], self.options),
         }
 
     async def run(
@@ -311,8 +319,9 @@ async def process_request(
     # Unused, kept for the stock signature.
     ctx: "Context | ExecutionContext | None",
     state: dict[str, Any] | None = None,
+    options: Options | None = None,
 ) -> tuple[js.Response, Future[None]]:
-    cycle = _HttpCycle(app)
+    cycle = _HttpCycle(app, options)
     await cycle.queue_body(req)
     request_task = create_task(cycle.run(req, env, state))
 
@@ -443,12 +452,16 @@ async def fetch(
     req: "Request | js.Request",
     env: Any,
     ctx: "Context | ExecutionContext | None" = None,
+    *,
+    options: Options | None = None,
 ) -> js.Response:
     logger.debug("ASGI request: %s %s", req.method, req.url)
     shutdown, state = await start_application(app)
 
     try:
-        result, request_task = await process_request(app, req, env, ctx, state=state)
+        result, request_task = await process_request(
+            app, req, env, ctx, state=state, options=options
+        )
     except Exception:
         logger.exception("ASGI request failed")
         await shutdown()
@@ -478,7 +491,7 @@ async def websocket(
     return await process_websocket(app, req, env, ctx)
 
 
-def entrypoint(app: Any) -> type[WorkerEntrypoint]:
+def entrypoint(app: Any, *, options: Options | None = None) -> type[WorkerEntrypoint]:
     """Create the default Worker entrypoint for an ASGI application."""
 
     class Default(WorkerEntrypoint):
@@ -487,7 +500,7 @@ async def fetch(self, request):
             ctx = getattr(self, "ctx", None)
             if (request.headers.get("upgrade") or "").lower() == "websocket":
                 return await websocket(app, request, self.env, ctx)
-            return await fetch(app, request, self.env, ctx)
+            return await fetch(app, request, self.env, ctx, options=options)
 
     return Default
 

workers/src/workers/gateway.py

@@ -0,0 +1,31 @@
+"""Options shared by the ASGI and WSGI adapters, as in the stock SDK at eec1e54."""
+
+from dataclasses import dataclass
+from typing import Any, Literal, get_args
+
+EncodeBody = Literal["automatic", "manual"]
+
+
+@dataclass(frozen=True, kw_only=True)
+class BaseOptions:
+    """Options shared by the ASGI and WSGI adapters.
+
+    Each adapter's `Options` subclasses this class.
+    """
+
+    # Whether the runtime encodes the body to match its `Content-Encoding`.
+    # https://developers.cloudflare.com/workers/runtime-apis/response/#the-encodebody-option
+    encode_body: EncodeBody = "automatic"
+
+    def __post_init__(self) -> None:
+        if self.encode_body not in get_args(EncodeBody):
+            raise ValueError(
+                f"encode_body must be one of {get_args(EncodeBody)}, "
+                f"got {self.encode_body!r}"
+            )
+
+    def response_init(self) -> dict[str, Any]:
+        """Extra keyword arguments for `Response.new`."""
+        if self.encode_body == "automatic":
+            return {}
+        return {"encodeBody": self.encode_body}

workers/src/workers/wsgi.py

@@ -9,6 +9,7 @@
 import logging
 import sys
 from collections.abc import Callable, Iterable, Iterator
+from dataclasses import dataclass
 from typing import Any
 from urllib.parse import SplitResult, unquote, urlsplit
 
@@ -18,12 +19,18 @@
 from ._adapter import NULL_BODY_STATUSES, body_encoding, request_headers
 from .context import ExecutionContext
 from .entrypoints import WorkerEntrypoint
+from .gateway import BaseOptions
 from .request import Request
 from .types import Context
 
 logger = logging.getLogger("wsgi")
 
 
+@dataclass(frozen=True, kw_only=True)
+class Options(BaseOptions):
+    """Options for the WSGI adapter."""
+
+
 def _wsgi_native_string(value: str) -> str:
     """Convert a (possibly non-ASCII) Python ``str`` into a WSGI "native string".
 
@@ -222,6 +229,7 @@ def _make_streaming_response(
     headers: "list[tuple[str, str]]",
     chunks: "Iterator[bytes]",
     on_close: "Callable[[], None]",
+    options: Options | None,
 ) -> js.Response:
     """Build a ``js.Response`` whose body is pulled lazily from *chunks*.
 
@@ -232,7 +240,7 @@ def _make_streaming_response(
     """
     from js import ReadableStream, Response
 
-    init = _response_init(status, headers)
+    init = _response_init(status, headers, options)
     if init["status"] in NULL_BODY_STATUSES:
         on_close()
         return Response.new(None, **init)
@@ -243,11 +251,13 @@ def _make_streaming_response(
     return Response.new(stream, **init)
 
 
-def _response_init(status: str, headers: "list[tuple[str, str]]") -> dict[str, Any]:
+def _response_init(
+    status: str, headers: "list[tuple[str, str]]", options: Options | None
+) -> dict[str, Any]:
     from js import Headers as JsHeaders
 
     code, _, reason = status.partition(" ")
-    init: dict[str, Any] = {"status": int(code), **body_encoding(headers)}
+    init: dict[str, Any] = {"status": int(code), **body_encoding(headers, options)}
     if reason:
         init["statusText"] = reason
 
@@ -338,6 +348,7 @@ def process_request(
     req: "Request | js.Request",
     env: Any,
     body: "bytes | io.IOBase",
+    options: Options | None = None,
 ) -> js.Response:
     environ = build_environ(req, env, body)
     start_response = _StartResponse()
@@ -371,7 +382,11 @@ def body_chunks() -> "Iterator[bytes]":
         yield from result_iter
 
     return _make_streaming_response(
-        start_response.status, start_response.headers, body_chunks(), close_all
+        start_response.status,
+        start_response.headers,
+        body_chunks(),
+        close_all,
+        options,
     )
 
 
@@ -380,24 +395,26 @@ async def fetch(
     req: "Request | js.Request",
     env: Any,
     ctx: "Context | ExecutionContext | None" = None,
+    *,
+    options: Options | None = None,
 ) -> js.Response:
     logger.debug("WSGI request: %s %s", req.method, req.url)
     body: bytes | io.IOBase | None = _make_wsgi_input(req)
     if body is None:
         body = await _read_body(req)
 
     try:
-        return process_request(app, req, env, body)
+        return process_request(app, req, env, body, options)
     except Exception:
         logger.exception("WSGI request failed")
         raise
 
 
-def entrypoint(app: Any) -> type[WorkerEntrypoint]:
+def entrypoint(app: Any, *, options: Options | None = None) -> type[WorkerEntrypoint]:
     """Create the default Worker entrypoint for a WSGI application."""
 
     class Default(WorkerEntrypoint):
         async def fetch(self, request):
-            return await fetch(app, request, self.env)
+            return await fetch(app, request, self.env, options=options)
 
     return Default

workers/tests/test_gateway.py

@@ -0,0 +1,24 @@
+import pytest
+
+from workers import asgi, wsgi
+from workers._adapter import body_encoding
+
+
+@pytest.mark.parametrize("adapter", [asgi, wsgi])
+def test_options_validate_encode_body(adapter) -> None:
+    assert adapter.Options().response_init() == {}
+    assert adapter.Options(encode_body="manual").response_init() == {
+        "encodeBody": "manual"
+    }
+    with pytest.raises(ValueError, match="encode_body"):
+        adapter.Options(encode_body="gzip")
+
+
+def test_explicit_options_override_content_encoding_detection() -> None:
+    gzipped = [("Content-Encoding", "gzip")]
+    assert body_encoding(gzipped, None) == {"encodeBody": "manual"}
+    assert body_encoding([], None) == {}
+    assert body_encoding(gzipped, asgi.Options()) == {}
+    assert body_encoding([], wsgi.Options(encode_body="manual")) == {
+        "encodeBody": "manual"
+    }