# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Error handling middleware for ASGI applications.
Catches exceptions raised during request processing and converts them
to appropriate HTTP responses. This is also the single seam where a 401
becomes a response, so the login challenge negotiation lives here.
Exception handling:
- Redirect: Returns 3xx redirect with Location header
- HTTPException: Returns status code with detail message; a 401 is
negotiated when the login surface is active (see below)
- Exception: Returns 500 Internal Server Error
Challenge negotiation (only when the server has ``login=True``):
A 401 is the point where the server asks the caller to authenticate.
API-first is preserved — a programmatic caller keeps the bare 401 (with
``WWW-Authenticate``) and now also gets a body ``{"login_url": ...}`` so an
SPA can find the page. A browser NAVIGATION (an http GET whose ``Accept``
includes ``text/html``) instead gets a 302 to the login page carrying the
original path+query as a validated ``next``. With the login surface off the
401 is emitted exactly as before (no body change, no redirect).
Config:
debug (bool): If True, include traceback in 500 responses. Default: False.
Note:
This middleware is enabled by default (middleware_default=True) and
runs early in the chain (middleware_order=100) to catch all errors.
Example:
Middleware is auto-enabled, but can be configured in the config.py
recipe::
root.middleware(errors={"debug": True}) # tracebacks in development
"""
from __future__ import annotations
import traceback
from typing import TYPE_CHECKING, Any
from urllib.parse import quote
from . import BaseMiddleware
from ..exceptions import HTTPException, Redirect
from ..utils import decode_headers, json_dumps, safe_next_path
if TYPE_CHECKING:
from ..types import ASGIApp, Receive, Scope, Send
LOGIN_URL = "/_server/login"
[docs]
class ErrorMiddleware(BaseMiddleware):
"""Error handling middleware for HTTP requests.
Wraps the application and catches exceptions, converting them to
appropriate HTTP error responses. Non-HTTP requests pass through unchanged.
Attributes:
debug: If True, include stack traces in 500 error responses.
Class Attributes:
middleware_name: "errors" - identifier for config.
middleware_order: 100 - runs early to catch all errors.
middleware_default: True - enabled by default.
"""
middleware_name = "errors"
middleware_order = 100
middleware_default = True
__slots__ = ("debug",)
[docs]
def __init__(
self,
app: ASGIApp,
debug: bool = False,
**kwargs: Any,
) -> None:
"""Initialize error middleware.
Args:
app: Next ASGI application in the middleware chain.
debug: Show tracebacks in 500 responses. Defaults to False.
**kwargs: Additional arguments passed to BaseMiddleware.
"""
super().__init__(app, **kwargs)
self.debug = debug
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Process request with error handling.
For HTTP requests, wraps the downstream app in try/except to catch
and handle exceptions. Non-HTTP requests (WebSocket, lifespan) pass
through without error handling.
Args:
scope: ASGI scope dictionary.
receive: ASGI receive callable.
send: ASGI send callable.
Note:
Exception priority: Redirect > HTTPException > generic Exception
"""
if scope["type"] != "http":
await self.app(scope, receive, send)
return
try:
await self.app(scope, receive, send)
except Redirect as e:
await self._send_redirect(send, e)
except HTTPException as e:
if e.status_code == 401 and self._login_active():
await self._send_challenge(scope, send, e)
else:
await self._send_http_error(send, e)
except Exception as e:
await self._send_server_error(send, e)
async def _send_redirect(self, send: Send, exc: Redirect) -> None:
"""Send HTTP redirect response.
Args:
send: ASGI send callable for response transmission.
exc: Redirect exception with target URL and status code.
Note:
Uses exc.status_code (default 307) and sets Location header.
Response body is empty.
"""
await send(
{
"type": "http.response.start",
"status": exc.status_code,
"headers": [(b"location", exc.url.encode())],
}
)
await send({"type": "http.response.body", "body": b""})
async def _send_http_error(self, send: Send, exc: HTTPException) -> None:
"""Send HTTP error response from HTTPException.
Args:
send: ASGI send callable for response transmission.
exc: HTTPException with status_code, detail, and optional headers.
Note:
Content-Type: text/plain; charset=utf-8
Body contains exc.detail message.
Additional headers from exc.headers are appended.
"""
body = exc.detail or ""
body_bytes = body.encode("utf-8")
headers: list[tuple[bytes, bytes]] = [
(b"content-type", b"text/plain; charset=utf-8"),
(b"content-length", str(len(body_bytes)).encode()),
]
if exc.headers:
headers.extend((k.encode(), v.encode()) for k, v in exc.headers)
await send({"type": "http.response.start", "status": exc.status_code, "headers": headers})
await send({"type": "http.response.body", "body": body_bytes})
def _login_active(self) -> bool:
"""True when the server carries an active login surface (``login=True``).
Used standalone (no server in the chain) the middleware has no login
surface, so the historical 401 is emitted unchanged.
"""
server = self.server
return bool(getattr(server, "login_enabled", False))
async def _send_challenge(self, scope: Scope, send: Send, exc: HTTPException) -> None:
"""Negotiate a 401 into a browser redirect or an API-friendly 401.
A browser navigation gets a 302 to the login page with the original
path+query as a validated ``next``; any other caller keeps the bare 401
(with its ``WWW-Authenticate``) and gains a ``{"login_url": ...}`` JSON
body so an SPA can drive the login itself.
"""
headers = scope.get("_headers") or decode_headers(scope)
if self._is_browser_navigation(scope, headers):
target = safe_next_path(self._original_target(scope))
location = f"{LOGIN_URL}?next={quote(target, safe='')}"
await self._send_redirect(send, Redirect(location))
else:
await self._send_api_challenge(send, exc)
def _is_browser_navigation(self, scope: Scope, headers: dict[str, str]) -> bool:
"""True for an http GET whose ``Accept`` asks for HTML (a navigation)."""
if scope.get("method", "").upper() != "GET":
return False
return "text/html" in (headers.get("accept") or "")
def _original_target(self, scope: Scope) -> str:
"""Rebuild the request's original path (+query) for the ``next`` value."""
path = scope.get("path", "/")
query = scope.get("query_string", b"")
query_str = query.decode("latin-1") if isinstance(query, bytes) else str(query)
return f"{path}?{query_str}" if query_str else path
async def _send_api_challenge(self, send: Send, exc: HTTPException) -> None:
"""Send the API 401: the bare status + WWW-Authenticate + login_url body."""
body_bytes = json_dumps({"login_url": LOGIN_URL})
headers: list[tuple[bytes, bytes]] = [
(b"content-type", b"application/json"),
(b"content-length", str(len(body_bytes)).encode()),
]
if exc.headers:
headers.extend((k.encode(), v.encode()) for k, v in exc.headers)
await send({"type": "http.response.start", "status": 401, "headers": headers})
await send({"type": "http.response.body", "body": body_bytes})
async def _send_server_error(self, send: Send, error: Exception) -> None:
"""Send 500 Internal Server Error response.
Args:
send: ASGI send callable for response transmission.
error: The unhandled exception that was caught.
Note:
If self.debug is True, includes full traceback in response body.
Otherwise, returns generic "Internal Server Error" message.
Content-Type: text/plain; charset=utf-8
"""
if self.debug:
body = f"Internal Server Error\n\n{traceback.format_exc()}"
else:
body = "Internal Server Error"
body_bytes = body.encode("utf-8")
await send(
{
"type": "http.response.start",
"status": 500,
"headers": [
(b"content-type", b"text/plain; charset=utf-8"),
(b"content-length", str(len(body_bytes)).encode()),
],
}
)
await send({"type": "http.response.body", "body": body_bytes})
if __name__ == "__main__":
pass