Surface non-Connect handler exceptions to user (#219)

Signed-off-by: Anuraag Agrawal <anuraaga@gmail.com>

Author: anuraaga

pyproject.toml

@@ -28,7 +28,7 @@ classifiers = [
   "Topic :: Software Development :: Libraries :: Python Modules",
   "Typing :: Typed",
 ]
-dependencies = ["protobuf>=5.28", "pyqwest>=0.4.1"]
+dependencies = ["protobuf>=5.28", "pyqwest>=0.5.1"]
 
 [project.urls]
 Documentation = "https://connectrpc.com/docs/python/getting-started/"

src/connectrpc/_server_async.py

@@ -232,7 +232,10 @@ async def __call__(
                     ctx,
                 )
         except Exception as e:
-            return await self._handle_error(e, ctx, send)
+            await self._handle_error(e, ctx, send)
+            if not isinstance(e, (ConnectError, HTTPException)):
+                raise
+            return None
 
         # Streams have their own error handling so move out of the try block.
         return await self._handle_stream(
@@ -486,6 +489,8 @@ async def _watch_for_disconnect() -> None:
                         "more_trailers": False,
                     }
                 )
+            if error and not isinstance(error, ConnectError):
+                raise error
 
     async def _handle_error(
         self, exc: Exception, ctx: RequestContext | None, send: ASGISendCallable

src/connectrpc/_server_sync.py

@@ -2,6 +2,7 @@
 
 import base64
 import functools
+import traceback
 from abc import ABC, abstractmethod
 from dataclasses import replace
 from http import HTTPStatus
@@ -46,9 +47,9 @@
     from .compression import Compression
 
     if sys.version_info >= (3, 11):
-        from wsgiref.types import StartResponse, WSGIEnvironment
+        from wsgiref.types import ErrorStream, StartResponse, WSGIEnvironment
     else:
-        from _typeshed.wsgi import StartResponse, WSGIEnvironment
+        from _typeshed.wsgi import ErrorStream, StartResponse, WSGIEnvironment
 else:
     StartResponse = "wsgiref.types.StartResponse"
     WSGIEnvironment = "wsgiref.types.WSGIEnvironment"
@@ -251,6 +252,7 @@ def __call__(
 
         except Exception as e:
             _drain_request_body(environ)
+            _maybe_log_exception(environ, e)
             return self._handle_error(e, ctx, start_response)
 
     def _handle_unary(
@@ -502,6 +504,7 @@ def _handle_stream(
             # response message will be handled by _response_stream, so here we have a
             # full error-only response.
             _drain_request_body(environ)
+            _maybe_log_exception(environ, e)
             _send_stream_response_headers(
                 start_response, protocol, codec, resp_compression.name(), ctx
             )
@@ -668,3 +671,12 @@ def _drain_request_body(environ: WSGIEnvironment) -> None:
         # server that doesn't do so, so we go ahead and do it ourselves.
         for _ in _read_body(environ):
             pass
+
+
+def _maybe_log_exception(environ: WSGIEnvironment, exc: Exception) -> None:
+    if isinstance(exc, (ConnectError, HTTPException)):
+        return
+    errors: ErrorStream = environ["wsgi.errors"]
+    errors.write(
+        f"Exception in WSGI application\n{''.join(traceback.format_exception(type(exc), exc, exc.__traceback__))}"
+    )

test/test_errors.py

@@ -3,7 +3,7 @@
 import asyncio
 import threading
 from http import HTTPStatus
-from typing import NoReturn
+from typing import TYPE_CHECKING, NoReturn
 
 import pytest
 from pyqwest import (
@@ -19,6 +19,7 @@
 )
 from pyqwest.testing import ASGITransport, WSGITransport
 
+from connectrpc._protocol import HTTPException
 from connectrpc.code import Code
 from connectrpc.errors import ConnectError
 
@@ -32,6 +33,9 @@
 )
 from .haberdasher_pb2 import Hat, Size
 
+if TYPE_CHECKING:
+    from connectrpc.request import RequestContext
+
 _errors = [
     (Code.CANCELED, "Operation was cancelled", 499),
     (Code.UNKNOWN, "An unknown error occurred", 500),
@@ -424,3 +428,148 @@ async def make_hat(self, request, ctx) -> NoReturn:
     assert exc_info.value.code == Code.DEADLINE_EXCEEDED
     assert exc_info.value.message == "Request timed out"
     assert recorded_timeout_header == "200"
+
+
+@pytest.mark.asyncio
+async def test_async_unhandled_exception_reraised() -> None:
+    class RaisingHaberdasher(Haberdasher):
+        async def make_hat(self, request, ctx) -> NoReturn:
+            raise TypeError("Something went wrong")
+
+    app = HaberdasherASGIApplication(RaisingHaberdasher())
+    transport = ASGITransport(app)
+    http_client = Client(transport)
+
+    async with HaberdasherClient(
+        "http://localhost", timeout_ms=200, http_client=http_client
+    ) as client:
+        with pytest.raises(ConnectError, match="Something went wrong"):
+            await client.make_hat(request=Size(inches=10))
+
+    assert isinstance(transport.app_exception, TypeError)
+    assert str(transport.app_exception) == "Something went wrong"
+
+
+@pytest.mark.asyncio
+async def test_async_unhandled_exception_reraised_stream() -> None:
+    class RaisingHaberdasher(Haberdasher):
+        def make_similar_hats(self, request: Size, ctx: RequestContext) -> NoReturn:
+            raise TypeError("Something went wrong")
+
+    app = HaberdasherASGIApplication(RaisingHaberdasher())
+    transport = ASGITransport(app)
+    http_client = Client(transport)
+
+    async with HaberdasherClient(
+        "http://localhost", timeout_ms=200, http_client=http_client
+    ) as client:
+        with pytest.raises(ConnectError, match="Something went wrong"):
+            async for _ in client.make_similar_hats(request=Size(inches=10)):
+                pass
+
+    assert isinstance(transport.app_exception, TypeError)
+    assert str(transport.app_exception) == "Something went wrong"
+
+
+@pytest.mark.asyncio
+async def test_async_connect_exception_not_reraised() -> None:
+    class RaisingHaberdasher(Haberdasher):
+        async def make_hat(self, request, ctx) -> NoReturn:
+            raise ConnectError(Code.INTERNAL, "We're broken")
+
+    app = HaberdasherASGIApplication(RaisingHaberdasher())
+    transport = ASGITransport(app)
+    http_client = Client(transport)
+
+    async with HaberdasherClient(
+        "http://localhost", timeout_ms=200, http_client=http_client
+    ) as client:
+        with pytest.raises(ConnectError, match="We're broken"):
+            await client.make_hat(request=Size(inches=10))
+
+    assert transport.app_exception is None
+
+
+@pytest.mark.asyncio
+async def test_async_connect_exception_not_reraised_stream() -> None:
+    class RaisingHaberdasher(Haberdasher):
+        def make_similar_hats(self, request: Size, ctx: RequestContext) -> NoReturn:
+            raise ConnectError(Code.INTERNAL, "We're broken")
+
+    app = HaberdasherASGIApplication(RaisingHaberdasher())
+    transport = ASGITransport(app)
+    http_client = Client(transport)
+
+    async with HaberdasherClient(
+        "http://localhost", timeout_ms=200, http_client=http_client
+    ) as client:
+        with pytest.raises(ConnectError, match="We're broken"):
+            async for _ in client.make_similar_hats(request=Size(inches=10)):
+                pass
+
+    assert transport.app_exception is None
+
+
+@pytest.mark.asyncio
+async def test_async_http_exception_not_reraised() -> None:
+    class RaisingHaberdasher(Haberdasher):
+        async def make_hat(self, request, ctx) -> NoReturn:
+            raise HTTPException(status=HTTPStatus.INTERNAL_SERVER_ERROR, headers=[])
+
+    app = HaberdasherASGIApplication(RaisingHaberdasher())
+    transport = ASGITransport(app)
+    http_client = Client(transport)
+
+    async with HaberdasherClient(
+        "http://localhost", timeout_ms=200, http_client=http_client
+    ) as client:
+        with pytest.raises(ConnectError, match="Internal Server Error"):
+            await client.make_hat(request=Size(inches=10))
+
+    assert transport.app_exception is None
+
+
+def test_sync_unhandled_exception_logged() -> None:
+    class RaisingHaberdasher(HaberdasherSync):
+        def make_hat(self, request, ctx) -> NoReturn:
+            raise TypeError("Something went wrong")
+
+    app = HaberdasherWSGIApplication(RaisingHaberdasher())
+    transport = WSGITransport(app)
+    http_client = SyncClient(transport)
+
+    with (
+        HaberdasherClientSync(
+            "http://localhost", timeout_ms=200, http_client=http_client
+        ) as client,
+        pytest.raises(ConnectError, match="Something went wrong"),
+    ):
+        client.make_hat(request=Size(inches=10))
+
+    logged_error = transport.error_stream.getvalue()
+    assert "Exception in WSGI application" in logged_error
+    assert "TypeError: Something went wrong" in logged_error
+    assert "Traceback" in logged_error
+
+
+def test_sync_unhandled_exception_logged_stream() -> None:
+    class RaisingHaberdasher(HaberdasherSync):
+        def make_similar_hats(self, request, ctx) -> NoReturn:
+            raise TypeError("Something went wrong")
+
+    app = HaberdasherWSGIApplication(RaisingHaberdasher())
+    transport = WSGITransport(app)
+    http_client = SyncClient(transport)
+
+    with (
+        HaberdasherClientSync(
+            "http://localhost", timeout_ms=200, http_client=http_client
+        ) as client,
+        pytest.raises(ConnectError, match="Something went wrong"),
+    ):
+        next(client.make_similar_hats(request=Size(inches=10)))
+
+    logged_error = transport.error_stream.getvalue()
+    assert "Exception in WSGI application" in logged_error
+    assert "TypeError: Something went wrong" in logged_error
+    assert "Traceback" in logged_error