🎨 web-server: exception handling framework

services/web/server/src/simcore_service_webserver/exception_handling/__init__.py

@@ -0,0 +1,12 @@
+from ._base import ExceptionHandlersMap, exception_handling_decorator
+from ._factory import ExceptionToHttpErrorMap, HttpErrorInfo, to_exceptions_handlers_map
+
+__all__: tuple[str, ...] = (
+    "ExceptionHandlersMap",
+    "ExceptionToHttpErrorMap",
+    "HttpErrorInfo",
+    "exception_handling_decorator",
+    "to_exceptions_handlers_map",
+)
+
+# nopycln: file

services/web/server/src/simcore_service_webserver/exception_handling/_base.py

@@ -0,0 +1,172 @@
+import functools
+import logging
+from collections.abc import Callable, Iterable
+from contextlib import AbstractAsyncContextManager
+from types import TracebackType
+from typing import Protocol, TypeAlias
+
+from aiohttp import web
+from servicelib.aiohttp.typing_extension import Handler as WebHandler
+from servicelib.aiohttp.typing_extension import Middleware as WebMiddleware
+
+_logger = logging.getLogger(__name__)
+
+
+class AiohttpExceptionHandler(Protocol):
+    __name__: str
+
+    async def __call__(
+        self,
+        request: web.Request,
+        exception: Exception,
+    ) -> web.StreamResponse:
+        """
+        Callback that handles an exception produced during a request and transforms it into a response
+
+        Arguments:
+            request -- current request
+            exception -- exception raised in web handler during this request
+        """
+
+
+ExceptionHandlersMap: TypeAlias = dict[type[Exception], AiohttpExceptionHandler]
+
+
+def _sort_exceptions_by_specificity(
+    exceptions: Iterable[type[Exception]], *, concrete_first: bool = True
+) -> list[type[Exception]]:
+    """
+    Keyword Arguments:
+        concrete_first -- If True, concrete subclasses precede their superclass (default: {True}).
+    """
+    return sorted(
+        exceptions,
+        key=lambda exc: sum(issubclass(e, exc) for e in exceptions if e is not exc),
+        reverse=not concrete_first,
+    )
+
+
+class ExceptionHandlingContextManager(AbstractAsyncContextManager):
+    """
+    A dynamic try-except context manager for handling exceptions in web handlers.
+    Maps exception types to corresponding handlers, allowing structured error management, i.e.
+    essentially something like
+    ```
+        try:
+
+            resp = await handler(request)
+
+        except exc_type1 as exc1:
+            resp = await exc_handler1(request)
+        except exc_type2 as exc1:
+            resp = await exc_handler2(request)
+        # etc
+
+    ```
+    and `exception_handlers_map` defines the mapping of exception types (`exc_type*`) to their handlers (`exc_handler*`).
+    """
+
+    def __init__(
+        self,
+        exception_handlers_map: ExceptionHandlersMap,
+        *,
+        request: web.Request,
+    ):
+        self._exc_handlers_map = exception_handlers_map
+        self._exc_types_by_specificity = _sort_exceptions_by_specificity(
+            list(self._exc_handlers_map.keys()), concrete_first=True
+        )
+        self._request: web.Request = request
+        self._response: web.StreamResponse | None = None
+
+    def _get_exc_handler_or_none(
+        self, exc_type: type[Exception], exc_value: Exception
+    ) -> AiohttpExceptionHandler | None:
+        exc_handler = self._exc_handlers_map.get(exc_type)
+        if not exc_handler and (
+            base_exc_type := next(
+                (
+                    _type
+                    for _type in self._exc_types_by_specificity
+                    if isinstance(exc_value, _type)
+                ),
+                None,
+            )
+        ):
+            exc_handler = self._exc_handlers_map[base_exc_type]
+        return exc_handler
+
+    async def __aenter__(self):
+        self._response = None
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> bool:
+        if (
+            exc_value is not None
+            and exc_type is not None
+            and isinstance(exc_value, Exception)
+            and issubclass(exc_type, Exception)
+            and (exc_handler := self._get_exc_handler_or_none(exc_type, exc_value))
+        ):
+            self._response = await exc_handler(
+                request=self._request, exception=exc_value
+            )
+            return True  # suppress
+        return False  # reraise
+
+    def get_response_or_none(self) -> web.StreamResponse | None:
+        """
+        Returns the response generated by the exception handler, if an exception was handled. Otherwise None
+        """
+        return self._response
+
+
+def exception_handling_decorator(
+    exception_handlers_map: dict[type[Exception], AiohttpExceptionHandler]
+) -> Callable[[WebHandler], WebHandler]:
+    """Creates a decorator to manage exceptions raised in a given route handler.
+    Ensures consistent exception management across decorated handlers.
+
+    SEE examples test_exception_handling
+    """
+
+    def _decorator(handler: WebHandler):
+        @functools.wraps(handler)
+        async def _wrapper(request: web.Request) -> web.StreamResponse:
+            cm = ExceptionHandlingContextManager(
+                exception_handlers_map, request=request
+            )
+            async with cm:
+                return await handler(request)
+
+            # If an exception was handled, return the exception handler's return value
+            response = cm.get_response_or_none()
+            assert response is not None  # nosec
+            return response
+
+        return _wrapper
+
+    return _decorator
+
+
+def exception_handling_middleware(
+    exception_handlers_map: dict[type[Exception], AiohttpExceptionHandler]
+) -> WebMiddleware:
+    """Constructs middleware to handle exceptions raised across app routes
+
+    SEE examples test_exception_handling
+    """
+    _handle_excs = exception_handling_decorator(
+        exception_handlers_map=exception_handlers_map
+    )
+
+    @web.middleware
+    async def middleware_handler(request: web.Request, handler: WebHandler):
+        return await _handle_excs(handler)(request)
+
+    return middleware_handler

services/web/server/src/simcore_service_webserver/exception_handling/_factory.py

@@ -0,0 +1,134 @@
+import logging
+from typing import NamedTuple, TypeAlias
+
+from aiohttp import web
+from common_library.error_codes import create_error_code
+from common_library.json_serialization import json_dumps
+from models_library.rest_error import ErrorGet
+from servicelib.aiohttp.web_exceptions_extension import get_all_aiohttp_http_exceptions
+from servicelib.logging_errors import create_troubleshotting_log_kwargs
+from servicelib.status_codes_utils import is_5xx_server_error, is_error
+
+from ._base import AiohttpExceptionHandler, ExceptionHandlersMap
+
+_logger = logging.getLogger(__name__)
+
+
+_STATUS_CODE_TO_HTTP_ERRORS: dict[
+    int, type[web.HTTPError]
+] = get_all_aiohttp_http_exceptions(web.HTTPError)
+
+
+class _DefaultDict(dict):
+    def __missing__(self, key):
+        return f"'{key}=?'"
+
+
+class HttpErrorInfo(NamedTuple):
+    """Info provided to auto-create HTTPError"""
+
+    status_code: int
+    msg_template: str  # sets HTTPError.reason
+
+
+ExceptionToHttpErrorMap: TypeAlias = dict[type[Exception], HttpErrorInfo]
+
+
+def create_error_response(error: ErrorGet, status_code: int) -> web.Response:
+    assert is_error(status_code), f"{status_code=} must be an error [{error=}]"  # nosec
+
+    return web.json_response(
+        data={"error": error.model_dump(exclude_unset=True, mode="json")},
+        dumps=json_dumps,
+        reason=error.message,
+        status=status_code,
+    )
+
+
+def create_exception_handler_from_http_info(
+    status_code: int,
+    msg_template: str,
+) -> AiohttpExceptionHandler:
+    """
+    Custom Exception-Handler factory
+
+    Creates a custom `WebApiExceptionHandler` that maps specific exception to a given http status code error
+
+    Given an `ExceptionToHttpErrorMap`, this function returns a handler that checks if an exception
+    matches one in the map, returning an HTTP error with the mapped status code and message.
+    Server errors (5xx) include additional logging with request context. Unmapped exceptions are
+    returned as-is for re-raising.
+
+    Arguments:
+        status_code: the http status code to associate at the web-api interface to this error
+        msg_template: a template string to pass to the HttpError
+
+    Returns:
+        A web api exception handler
+    """
+    assert is_error(  # nosec
+        status_code
+    ), f"{status_code=} must be an error [{msg_template=}]"
+
+    async def _exception_handler(
+        request: web.Request,
+        exception: BaseException,
+    ) -> web.Response:
+
+        # safe formatting, i.e. does not raise
+        user_msg = msg_template.format_map(
+            _DefaultDict(getattr(exception, "__dict__", {}))
+        )
+
+        error = ErrorGet.model_construct(message=user_msg)
+
+        if is_5xx_server_error(status_code):
+            oec = create_error_code(exception)
+            _logger.exception(
+                **create_troubleshotting_log_kwargs(
+                    user_msg,
+                    error=exception,
+                    error_code=oec,
+                    error_context={
+                        "request": request,
+                        "request.remote": f"{request.remote}",
+                        "request.method": f"{request.method}",
+                        "request.path": f"{request.path}",
+                    },
+                )
+            )
+            error = ErrorGet.model_construct(message=user_msg, support_id=oec)
+
+        return create_error_response(error, status_code=status_code)
+
+    return _exception_handler
+
+
+def to_exceptions_handlers_map(
+    exc_to_http_error_map: ExceptionToHttpErrorMap,
+) -> ExceptionHandlersMap:
+    """Data adapter to convert ExceptionToHttpErrorMap ot ExceptionHandlersMap, i.e.
+    - from  { exc_type: (status, msg), ... }
+    - to   { exc_type: callable, ... }
+    """
+    exc_handlers_map: ExceptionHandlersMap = {
+        exc_type: create_exception_handler_from_http_info(
+            status_code=info.status_code, msg_template=info.msg_template
+        )
+        for exc_type, info in exc_to_http_error_map.items()
+    }
+
+    return exc_handlers_map
+
+
+def create_http_error_exception_handlers_map() -> ExceptionHandlersMap:
+    """
+    Auto create handlers for **all** web.HTTPError
+    """
+    exc_handlers_map: ExceptionHandlersMap = {
+        exc_type: create_exception_handler_from_http_info(
+            status_code=code, msg_template="{reason}"
+        )
+        for code, exc_type in _STATUS_CODE_TO_HTTP_ERRORS.items()
+    }
+    return exc_handlers_map