Coverage for src/lexigram/web/filters/builtin.py: 24%

1"""Built-in exception filters for Lexigram Web.""" 

2 

3from __future__ import annotations 

4 

5import dataclasses 

6from typing import Any 

7 

8from lexigram.web.errors.html_error_renderer import DebugHtmlErrorRenderer 

9from lexigram.web.exceptions import ( 

10 DependencyResolutionError, 

11 HTTPError, 

12) 

13from lexigram.web.transport.responses import JSONResponse 

14 

15 

16class ValidationErrorFilter: 

17 """Filter for handling Pydantic and Lexigram validation errors.""" 

18 

19 def can_handle(self, exc: Exception) -> bool: 

20 """Check if exception is a validation error.""" 

21 name = exc.__class__.__name__ 

22 return ( 

23 name in ("ValidationError", "PydanticValidationError") 

24 or getattr(exc, "code", None) in ("validation_error", "VALIDATION_ERROR") 

25 or ( 

26 isinstance(exc, ValueError) 

27 and any(kw in str(exc) for kw in ("validation", "Invalid", "coercion")) 

28 ) 

29 ) 

30 

31 async def handle(self, exc: Any, _request: Any) -> JSONResponse: 

32 """Transform ValidationError into a 422 JSONResponse with RFC 7807 format.""" 

33 raw_errors: list[Any] = [] 

34 if hasattr(exc, "errors") and callable(exc.errors): 

35 raw_errors = exc.errors() 

36 elif hasattr(exc, "errors") and isinstance(exc.errors, list): 

37 raw_errors = exc.errors 

38 elif hasattr(exc, "args") and exc.args: 

39 raw_errors = [{"msg": str(arg)} for arg in exc.args] 

40 

41 errors = [] 

42 for e in raw_errors: 

43 if dataclasses.is_dataclass(e) and not isinstance(e, type): 

44 errors.append(dataclasses.asdict(e)) 

45 elif isinstance(e, dict): 

46 errors.append(e) 

47 elif hasattr(e, "field") and hasattr(e, "message"): 

48 errors.append( 

49 { 

50 "field": e.field, 

51 "message": e.message, 

52 "code": getattr(e, "code", None), 

53 } 

54 ) 

55 else: 

56 errors.append({"msg": str(e)}) 

57 

58 from lexigram.web.errors.problem_detail import ProblemDetail 

59 

60 pd = ProblemDetail( 

61 type="urn:lexigram:validation-error", 

62 title="Validation Error", 

63 status=422, 

64 detail="Request validation failed", 

65 errors=errors, 

66 ) 

67 return JSONResponse( 

68 content=pd.to_dict(), 

69 status_code=422, 

70 media_type="application/problem+json", 

71 ) 

72 

73 

74class DependencyResolutionFilter: 

75 """Filter for handling DependencyResolutionError.""" 

76 

77 def can_handle(self, exc: Exception) -> bool: 

78 """Check if exception is a DependencyResolutionError.""" 

79 return ( 

80 isinstance(exc, DependencyResolutionError) 

81 or exc.__class__.__name__ == "DependencyResolutionError" 

82 or getattr(exc, "code", None) == "dependency_resolution_error" 

83 ) 

84 

85 async def handle(self, exc: Any, _request: Any) -> JSONResponse: 

86 """Transform DependencyResolutionError into a 500 RFC 7807 JSONResponse.""" 

87 from lexigram.web.errors.problem_detail import ProblemDetail 

88 

89 pd = ProblemDetail.internal_error(detail=getattr(exc, "detail", str(exc))) 

90 return JSONResponse( 

91 content=pd.to_dict(), 

92 status_code=500, 

93 media_type="application/problem+json", 

94 ) 

95 

96 

97class DefaultExceptionFilter: 

98 """Default filter that handles HTTPError and generic exceptions. 

99 

100 When ``debug=True`` and the HTTP client sends ``Accept: text/html``, 

101 the response is a rich HTML error page showing the traceback and request 

102 details. In all other cases a structured JSON response is returned. 

103 """ 

104 

105 def __init__(self, debug: bool = False) -> None: 

106 """Initialise the filter. 

107 

108 Args: 

109 debug: When ``True``, HTML error pages are rendered for browser 

110 clients. Set this based on ``ServerConfig.debug`` or 

111 the ``LEX_DEBUG`` environment variable. 

112 """ 

113 self._debug = debug 

114 self._html_renderer = DebugHtmlErrorRenderer() 

115 

116 def can_handle(self, exc: Exception) -> bool: 

117 """Handle HTTPErrors and LexigramErrors specifically.""" 

118 from lexigram.contracts.exceptions import LexigramError 

119 

120 return ( 

121 isinstance(exc, (HTTPError, LexigramError)) 

122 or getattr(exc, "status_code", None) is not None 

123 or hasattr(exc, "code") 

124 ) 

125 

126 async def handle(self, exc: Exception, request: Any) -> Any: 

127 """Transform the exception into a JSON or HTML response. 

128 

129 When ``debug=True`` and the request prefers HTML, a rich debug page 

130 is returned instead of JSON. All JSON responses use RFC 7807 

131 ``application/problem+json`` format. 

132 """ 

133 from lexigram.web.errors.problem_detail import ProblemDetail 

134 

135 if isinstance(exc, HTTPError): 

136 if self._debug and self._html_renderer.should_render(request): 

137 return self._html_renderer.render( 

138 exc, 

139 request, 

140 status_code=exc.status_code, 

141 title=f"{exc.status_code} {exc.detail}", 

142 ) 

143 code = getattr(exc, "code", None) 

144 pd = ProblemDetail( 

145 type=( 

146 f"urn:lexigram:{code.lower().replace('_', '-')}" 

147 if code 

148 else "about:blank" 

149 ), 

150 title=exc.detail, 

151 status=exc.status_code, 

152 detail=exc.detail, 

153 ) 

154 return JSONResponse( 

155 content=pd.to_dict(), 

156 status_code=exc.status_code, 

157 headers=exc.headers, 

158 media_type="application/problem+json", 

159 ) 

160 

161 # Handle domain errors via MRO-ordered type mapping. 

162 from lexigram.contracts.exceptions.domain import ( 

163 AuthenticationError, 

164 AuthorizationError, 

165 ConflictError, 

166 DomainError, 

167 NotFoundError, 

168 PermissionDeniedError, 

169 ) 

170 

171 status_code = 500 

172 type_uri = "urn:lexigram:internal-error" 

173 title = "Internal Server Error" 

174 

175 if isinstance(exc, NotFoundError): 

176 status_code, type_uri, title = 404, "urn:lexigram:not-found", "Not Found" 

177 elif isinstance(exc, (PermissionDeniedError, AuthorizationError)): 

178 status_code, type_uri, title = 403, "urn:lexigram:forbidden", "Forbidden" 

179 elif isinstance(exc, AuthenticationError): 

180 status_code, type_uri, title = ( 

181 401, 

182 "urn:lexigram:unauthorized", 

183 "Unauthorized", 

184 ) 

185 elif isinstance(exc, ConflictError): 

186 status_code, type_uri, title = 409, "urn:lexigram:conflict", "Conflict" 

187 elif isinstance(exc, DomainError): 

188 status_code, type_uri, title = ( 

189 400, 

190 "urn:lexigram:bad-request", 

191 "Bad Request", 

192 ) 

193 

194 if self._debug and self._html_renderer.should_render(request): 

195 return self._html_renderer.render(exc, request, status_code=status_code) 

196 

197 pd = ProblemDetail( 

198 type=type_uri, 

199 title=title, 

200 status=status_code, 

201 detail=str(exc), 

202 ) 

203 return JSONResponse( 

204 content=pd.to_dict(), 

205 status_code=status_code, 

206 media_type="application/problem+json", 

207 ) 

208 

209 

210__all__ = [ 

211 "DefaultExceptionFilter", 

212 "DependencyResolutionFilter", 

213 "ValidationErrorFilter", 

214]