Replace current WSGIMiddleware implementation by a2wsgi one

docs/settings.md

@@ -80,6 +80,10 @@ Using Uvicorn with watchfiles will enable the following options (which are other
 Note that WSGI mode always disables WebSocket support, as it is not supported by the WSGI interface.
 **Options:** *'auto', 'asgi3', 'asgi2', 'wsgi'.* **Default:** *'auto'*.
 
+!!! warning
+    Uvicorn's native WSGI implementation is deprecated, you should switch
+    to [a2wsgi](https://github.com/abersheeran/a2wsgi) (`pip install a2wsgi`).
+
 ## HTTP
 
 * `--root-path <str>` - Set the ASGI `root_path` for applications submounted below a given URL path.

requirements.txt

@@ -11,6 +11,7 @@ build==0.9.0
 twine==4.0.1
 
 # Testing
+a2wsgi==1.6.0
 autoflake==1.7.7
 black==22.10.0
 flake8==3.9.2

setup.cfg

@@ -34,6 +34,7 @@ filterwarnings=
     # Turn warnings that aren't filtered into exceptions
     error
     ignore: \"watchgod\" is depreciated\, you should switch to watchfiles \(`pip install watchfiles`\)\.:DeprecationWarning
+    ignore: Uvicorn's native WSGI implementation is deprecated, you should switch to a2wsgi \(`pip install a2wsgi`\)\.:DeprecationWarning
     ignore: 'cgi' is deprecated and slated for removal in Python 3\.13:DeprecationWarning
 
 [coverage:run]

tests/middleware/test_wsgi.py

@@ -1,12 +1,13 @@
 import io
 import sys
-from typing import TYPE_CHECKING, AsyncGenerator, List
+from typing import TYPE_CHECKING, AsyncGenerator, Callable, List
 
+import a2wsgi
 import httpx
 import pytest
 
 from uvicorn._types import Environ, StartResponse
-from uvicorn.middleware.wsgi import WSGIMiddleware, build_environ
+from uvicorn.middleware import wsgi
 
 if TYPE_CHECKING:
     from asgiref.typing import HTTPRequestEvent, HTTPScope
@@ -34,7 +35,7 @@ def echo_body(environ: Environ, start_response: StartResponse) -> List[bytes]:
     return [output]
 
 
-def raise_exception(environ: Environ, start_response: StartResponse) -> RuntimeError:
+def raise_exception(environ: Environ, start_response: StartResponse) -> List[bytes]:
     raise RuntimeError("Something went wrong")
 
 
@@ -52,57 +53,62 @@ def return_exc_info(environ: Environ, start_response: StartResponse) -> List[byt
         return [output]
 
 
+@pytest.fixture(params=[wsgi._WSGIMiddleware, a2wsgi.WSGIMiddleware])
+def wsgi_middleware(request: pytest.FixtureRequest) -> Callable:
+    return request.param
+
+
 @pytest.mark.anyio
-async def test_wsgi_get() -> None:
-    app = WSGIMiddleware(hello_world)
+async def test_wsgi_get(wsgi_middleware: Callable) -> None:
+    app = wsgi_middleware(hello_world)
     async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
         response = await client.get("/")
     assert response.status_code == 200
     assert response.text == "Hello World!\n"
 
 
 @pytest.mark.anyio
-async def test_wsgi_post() -> None:
-    app = WSGIMiddleware(echo_body)
+async def test_wsgi_post(wsgi_middleware: Callable) -> None:
+    app = wsgi_middleware(echo_body)
     async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
         response = await client.post("/", json={"example": 123})
     assert response.status_code == 200
     assert response.text == '{"example": 123}'
 
 
 @pytest.mark.anyio
-async def test_wsgi_put_more_body() -> None:
+async def test_wsgi_put_more_body(wsgi_middleware: Callable) -> None:
     async def generate_body() -> AsyncGenerator[bytes, None]:
         for _ in range(1024):
             yield b"123456789abcdef\n" * 64
 
-    app = WSGIMiddleware(echo_body)
+    app = wsgi_middleware(echo_body)
     async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
         response = await client.put("/", content=generate_body())
     assert response.status_code == 200
     assert response.text == "123456789abcdef\n" * 64 * 1024
 
 
 @pytest.mark.anyio
-async def test_wsgi_exception() -> None:
+async def test_wsgi_exception(wsgi_middleware: Callable) -> None:
     # Note that we're testing the WSGI app directly here.
     # The HTTP protocol implementations would catch this error and return 500.
-    app = WSGIMiddleware(raise_exception)
+    app = wsgi_middleware(raise_exception)
     async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
         with pytest.raises(RuntimeError):
             await client.get("/")
 
 
 @pytest.mark.anyio
-async def test_wsgi_exc_info() -> None:
+async def test_wsgi_exc_info(wsgi_middleware: Callable) -> None:
     # Note that we're testing the WSGI app directly here.
     # The HTTP protocol implementations would catch this error and return 500.
-    app = WSGIMiddleware(return_exc_info)
+    app = wsgi_middleware(return_exc_info)
     async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
         with pytest.raises(RuntimeError):
             response = await client.get("/")
 
-    app = WSGIMiddleware(return_exc_info)
+    app = wsgi_middleware(return_exc_info)
     transport = httpx.ASGITransport(
         app=app,
         raise_app_exceptions=False,
@@ -136,6 +142,6 @@ def test_build_environ_encoding() -> None:
         "body": b"",
         "more_body": False,
     }
-    environ = build_environ(scope, message, io.BytesIO(b""))
+    environ = wsgi.build_environ(scope, message, io.BytesIO(b""))
     assert environ["PATH_INFO"] == "/文".encode("utf8").decode("latin-1")
     assert environ["HTTP_KEY"] == "value1,value2"

uvicorn/middleware/wsgi.py

@@ -2,6 +2,7 @@
 import concurrent.futures
 import io
 import sys
+import warnings
 from collections import deque
 from typing import TYPE_CHECKING, Deque, Iterable, Optional, Tuple
 
@@ -73,8 +74,13 @@ def build_environ(
     return environ
 
 
-class WSGIMiddleware:
+class _WSGIMiddleware:
     def __init__(self, app: WSGIApp, workers: int = 10):
+        warnings.warn(
+            "Uvicorn's native WSGI implementation is deprecated, you "
+            "should switch to a2wsgi (`pip install a2wsgi`).",
+            DeprecationWarning,
+        )
         self.app = app
         self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=workers)
 
@@ -188,3 +194,9 @@ def wsgi(self, environ: Environ, start_response: StartResponse) -> None:
         }
         self.send_queue.append(empty_body)
         self.loop.call_soon_threadsafe(self.send_event.set)
+
+
+try:
+    from a2wsgi import WSGIMiddleware
+except ModuleNotFoundError:
+    WSGIMiddleware = _WSGIMiddleware  # type: ignore[misc, assignment]