1r"""
2================
3Error Middleware
4================
5
6.. versionadded:: 0.2.0
7
8Middleware to handle errors in aiohttp applications.
9
10.. versionchanged:: 1.0.0
11
12Previosly, ``error_middleware`` required ``default_handler`` to be passed
13on initialization. However in **1.0.0** version ``aiohttp-middlewares`` ships
14default error handler, which log exception traceback into
15``aiohttp_middlewares.error`` logger and responds with given JSON:
16
17.. code-block:: json
18
19    {
20        "detail": "str"
21    }
22
23For example, if view handler raises ``ValueError("wrong value")`` the default
24error handler provides 500 Server Error JSON:
25
26.. code-block:: json
27
28    {
29        "detail": "wrong value"
30    }
31
32In same time, it is still able to provide custom default error handler if you
33need more control on error handling.
34
35Other notable change in **1.0.0** version is allowing to ignore exception or
36tuple of exceptions (as in ``try/catch`` block) from handling via middleware.
37This might be helpful, when you don't want, for example, to have in Sentry
38``web.HTTPNotFound`` and/or ``web.BadRequest`` errors.
39
40Usage
41=====
42
43.. code-block:: python
44
45    import re
46
47    from aiohttp import web
48    from aiohttp_middlewares import (
49        default_error_handler,
50        error_context,
51        error_middleware,
52    )
53
54    # Error handler for API requests
55    async def api_error(request: web.Request) -> web.Response:
56        with error_context(request) as context:
57            return web.json_response(
58                context.data, status=context.status
59            )
60
61
62    # Basic usage (default error handler for whole application)
63    app = web.Application(middlewares=[error_middleware()])
64
65    # Advanced usage (multiple error handlers for different
66    # application parts)
67    app = web.Application(
68        middlewares=[
69            error_middleware(
70                default_handler=default_error_handler,
71                config={re.compile(r"^\/api"): api_error},
72            )
73        ]
74    )
75
76    # Ignore aiohttp.web HTTP Not Found errors from handling via middleware
77    app = web.Application(
78        middlewares=[
79            error_middleware(ignore_exceptions=web.HTTPNotFound)
80        ]
81    )
82
83"""
84
85import logging
86from contextlib import contextmanager
87from functools import partial
88from typing import Dict, Iterator, Optional, Tuple, Union
89
90import attr
91from aiohttp import web
92
93from aiohttp_middlewares.annotations import (
94    DictStrAny,
95    ExceptionType,
96    Handler,
97    Middleware,
98    Url,
99)
100from aiohttp_middlewares.utils import match_path
101
102
103DEFAULT_EXCEPTION = Exception("Unhandled aiohttp-middlewares exception.")
104REQUEST_ERROR_KEY = "error"
105
106Config = Dict[Url, Handler]
107logger = logging.getLogger(__name__)
108
109
110@attr.dataclass(frozen=True, slots=True)
111class ErrorContext:
112    """Context with all necessary data about the error."""
113
114    err: Exception
115    message: str
116    status: int
117    data: DictStrAny
118
119
120async def default_error_handler(request: web.Request) -> web.Response:
121    """Default error handler to respond with JSON error details.
122
123    If, for example, ``aiohttp.web`` view handler raises
124    ``ValueError("wrong value")`` exception, default error handler will produce
125    JSON response of 500 status with given content:
126
127    .. code-block:: json
128
129        {
130            "detail": "wrong value"
131        }
132
133    And to see the whole exception traceback in logs you need to enable
134    ``aiohttp_middlewares`` in logging config.
135
136    .. versionadded:: 1.0.0
137    """
138    with error_context(request) as context:
139        logger.error(context.message, exc_info=True)
140        return web.json_response(context.data, status=context.status)
141
142
143@contextmanager
144def error_context(request: web.Request) -> Iterator[ErrorContext]:
145    """Context manager to retrieve error data inside of error handler (view).
146
147    The result instance will contain:
148
149    - Error itself
150    - Error message (by default: ``str(err)``)
151    - Error status (by default: ``500``)
152    - Error data dict (by default: ``{"detail": str(err)}``)
153    """
154    err = get_error_from_request(request)
155
156    message = getattr(err, "message", None) or str(err)
157    data = getattr(err, "data", None) or {"detail": message}
158    status = getattr(err, "status", None) or 500
159
160    yield ErrorContext(err=err, message=message, status=status, data=data)
161
162
163def error_middleware(
164    *,
165    default_handler: Handler = default_error_handler,
166    config: Config = None,
167    ignore_exceptions: Union[ExceptionType, Tuple[ExceptionType, ...]] = None
168) -> Middleware:
169    """Middleware to handle exceptions in aiohttp applications.
170
171    To catch all possible errors, please put this middleware on top of your
172    ``middlewares`` list (**but after CORS middleware if it is used**) as:
173
174    .. code-block:: python
175
176        from aiohttp import web
177        from aiohttp_middlewares import (
178            error_middleware,
179            timeout_middleware,
180        )
181
182        app = web.Application(
183            midllewares=[error_middleware(...), timeout_middleware(...)]
184        )
185
186    :param default_handler:
187        Default handler to called on error catched by error middleware.
188    :param config:
189        When application requires multiple error handlers, provide mapping in
190        format ``Dict[Url, Handler]``, where ``Url`` can be an exact string
191        to match path or regex and ``Handler`` is a handler to be called when
192        ``Url`` matches current request path if any.
193    :param ignore_exceptions:
194        Do not process given exceptions via error middleware.
195    """
196    get_response = partial(
197        get_error_response,
198        default_handler=default_handler,
199        config=config,
200        ignore_exceptions=ignore_exceptions,
201    )
202
203    @web.middleware
204    async def middleware(
205        request: web.Request, handler: Handler
206    ) -> web.StreamResponse:
207        try:
208            return await handler(request)
209        except Exception as err:  # noqa: PIE786
210            return await get_response(request, err)
211
212    return middleware
213
214
215def get_error_from_request(request: web.Request) -> Exception:
216    """Get previously stored error from request dict.
217
218    Return default exception if nothing stored before.
219    """
220    return request.get(REQUEST_ERROR_KEY) or DEFAULT_EXCEPTION
221
222
223def get_error_handler(
224    request: web.Request, config: Optional[Config]
225) -> Optional[Handler]:
226    """Find error handler matching current request path if any."""
227    if not config:
228        return None
229
230    path = request.rel_url.path
231    for item, handler in config.items():
232        if match_path(item, path):
233            return handler
234
235    return None
236
237
238async def get_error_response(
239    request: web.Request,
240    err: Exception,
241    *,
242    default_handler: Handler = default_error_handler,
243    config: Config = None,
244    ignore_exceptions: Union[ExceptionType, Tuple[ExceptionType, ...]] = None
245) -> web.StreamResponse:
246    """Actual coroutine to get response for given request & error.
247
248    .. versionadded:: 1.1.0
249
250    This is a cornerstone of error middleware and can be reused in attempt to
251    overwrite error middleware logic.
252
253    For example, when you need to post-process response and it may result in
254    extra exceptions it is useful to make ``custom_error_middleware`` as
255    follows,
256
257    .. code-block:: python
258
259        from aiohttp import web
260        from aiohttp_middlewares import get_error_response
261        from aiohttp_middlewares.annotations import Handler
262
263
264        @web.middleware
265        async def custom_error_middleware(
266            request: web.Request, handler: Handler
267        ) -> web.StreamResponse:
268            try:
269                response = await handler(request)
270                post_process_response(response)
271            except Exception as err:
272                return await get_error_response(request, err)
273    """
274    if ignore_exceptions and isinstance(err, ignore_exceptions):
275        raise err
276
277    set_error_to_request(request, err)
278    error_handler = get_error_handler(request, config) or default_handler
279    return await error_handler(request)
280
281
282def set_error_to_request(request: web.Request, err: Exception) -> Exception:
283    """Store catched error to request dict."""
284    request[REQUEST_ERROR_KEY] = err
285    return err
286