compiling old ideas

services/web/server/src/simcore_service_webserver/exceptions_handlers.py

@@ -1,8 +1,9 @@
 import functools
 import logging
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable, MutableMapping
 from contextlib import contextmanager
-from typing import NamedTuple, Protocol, TypeAlias
+from http import HTTPStatus
+from typing import Any, NamedTuple, Protocol, TypeAlias, cast
 
 from aiohttp import web
 from servicelib.aiohttp.typing_extension import Handler
@@ -12,6 +13,10 @@
 
 _logger = logging.getLogger(__name__)
 
+#
+# Definition
+#
+
 
 class WebApiExceptionHandler(Protocol):
     def __call__(
@@ -189,3 +194,128 @@ def _exception_handler(
         return exception
 
     return _exception_handler
+
+
+# OLD VERSION OF IT with really good ideas ! --------------------
+
+
+#
+# Defines exception handler as somethign that returns responses, as fastapi, and not new exceptions!
+# in reality this can be reinterpreted in aiohttp since all responses can be represented as exceptions.
+# Not true because fastapi.HTTPException does actually the same! Better responses because this weay we do not
+# need return None or the exception itself which as we saw in the tests, it causes troubles!
+#
+
+ExceptionHandler: TypeAlias = Callable[
+    [web.Request, Exception], Awaitable[web.Response]
+]
+
+ExceptionsMap: TypeAlias = dict[type[Exception], type[web.HTTPException]]
+
+ExceptionHandlerRegistry: TypeAlias = dict[type[Exception], ExceptionHandler]
+
+
+# injects the exceptions in a scope, e.g. an app state or some container like routes/ but to use in a module only e.g.
+# as decorator or context manager?
+
+
+def setup_exception_handlers(scope: MutableMapping[str, Any]):
+    scope["exceptions_handlers"] = {}
+    scope["exceptions_map"] = {}
+    # but this is very specific because it responds with only status! you migh want to have different
+    # type of bodies, etc
+
+
+def _get_exception_handler_registry(
+    scope: MutableMapping[str, Any]
+) -> ExceptionHandlerRegistry:
+    return scope.get("exceptions_handlers", {})
+
+
+def add_exception_handler(
+    scope: MutableMapping[str, Any],
+    exc_class: type[Exception],
+    handler: ExceptionHandler,
+):
+    scope["exceptions_handlers"][exc_class] = handler
+
+
+def _create_exception_handler_mapper(
+    exc_class: type[Exception],
+    http_exc_class: type[web.HTTPException],
+) -> ExceptionHandler:
+    error_code = f"{exc_class.__name__}"  # status_code.error_code
+
+    async def _exception_handler(_: web.Request, exc: Exception) -> web.Response:
+        # TODO: a better way to add error_code. TODO: create the envelope!
+        return http_exc_class(reason=f"{exc} [{error_code}]")
+
+    return _exception_handler
+
+
+def add_exception_mapper(
+    scope: MutableMapping[str, Any],
+    exc_class: type[Exception],
+    http_exc_class: type[web.HTTPException],
+):
+    # adds exception handler to scope
+    scope["exceptions_map"][exc_class] = http_exc_class
+    add_exception_handler(
+        scope,
+        exc_class,
+        handler=_create_exception_handler_mapper(exc_class, http_exc_class),
+    )
+
+
+async def handle_request_with_exception_handling_in_scope(
+    handler: Handler,
+    request: web.Request,
+    scope: MutableMapping[str, Any] | None = None,
+) -> web.Response:
+    try:
+        resp = await handler(request)
+        return cast(web.Response, resp)
+
+    except Exception as exc:  # pylint: disable=broad-exception-caught
+        scope = scope or request.app
+        if exception_handler := _get_exception_handler_registry(scope).get(
+            type(exc), None
+        ):
+            resp = await exception_handler(request, exc)
+        else:
+            resp = web.HTTPInternalServerError()
+
+        if isinstance(resp, web.HTTPError):
+            # NOTE: this should not happen anymore! as far as I understand!?
+            raise resp from exc
+        return resp
+
+
+def handle_registered_exceptions(scope: MutableMapping[str, Any] | None = None):
+    def _decorator(handler: Handler):
+        @functools.wraps(handler)
+        async def _wrapper(request: web.Request) -> web.Response:
+            return await handle_request_with_exception_handling_in_scope(
+                handler, request, scope
+            )
+
+        return _wrapper
+
+    return _decorator
+
+
+# If I have all the status codes mapped, I can definitively use that info to create `responses`
+# for fastapi to render the OAS preoperly
+def openapi_error_responses(
+    exceptions_map: ExceptionsMap,
+) -> dict[HTTPStatus, dict[str, Any]]:
+    responses = {}
+
+    for exc_class, http_exc_class in exceptions_map.items():
+        status_code = HTTPStatus(http_exc_class.status_code)
+        if status_code not in responses:
+            responses[status_code] = {"description": f"{exc_class.__name__}"}
+        else:
+            responses[status_code]["description"] += f", {exc_class.__name__}"
+
+    return responses

services/web/server/tests/unit/isolated/test_exceptions_handlers.py

@@ -11,6 +11,12 @@
 from aiohttp.test_utils import make_mocked_request
 from servicelib.aiohttp import status
 from simcore_service_webserver.errors import WebServerBaseError
+from simcore_service_webserver.exception_handlers import (
+    add_exception_handler,
+    add_exception_mapper,
+    handle_registered_exceptions,
+    setup_exception_handlers,
+)
 from simcore_service_webserver.exceptions_handlers import (
     HttpErrorInfo,
     _handled_exception_context,
@@ -146,3 +152,32 @@ async def _rest_handler(request: web.Request):
         resp = await _rest_handler(
             make_mocked_request("GET", "/foo?raise=ArithmeticError")
         )
+
+
+def test_it():
+    class MyException(Exception):
+        ...
+
+    class OtherException(Exception):
+        ...
+
+    app = web.Application()
+
+    async def my_error_handler(request: web.Request, exc: MyException):
+        return web.HTTPNotFound()
+
+    # define at the app level
+    setup_exception_handlers(app)
+    add_exception_handler(app, MyException, my_error_handler)
+    add_exception_mapper(app, OtherException, web.HTTPNotFound)
+
+    async def foo():
+        raise MyException
+
+    routes = web.RouteTableDef()
+
+    @routes.get("/home")
+    @handle_registered_exceptions()
+    async def home(_request: web.Request):
+        await foo()
+        return web.HTTPOk()