Coverage for src/lexigram/web/routing/result_bridge.py: 26%

1"""Result-to-HTTP response mapper. 

2 

3Converts ``Result[T, E]`` domain errors to appropriate HTTP status codes, 

4allowing controller methods to return typed domain errors without manual 

5HTTP mapping. 

6 

7Usage:: 

8 

9 # Register a custom mapping 

10 ResultResponseMapper.register(MyConflictError, 409) 

11 

12 # In a controller (handled automatically by the serialization pipeline): 

13 @get("/{user_id}") 

14 async def get_user(self, user_id: str) -> Result[User, NotFoundError]: 

15 return await self.service.find(user_id) 

16 # If Err(NotFoundError(...)) → HTTP 404 

17 # If Ok(user) → HTTP 200 with serialized user 

18""" 

19 

20from __future__ import annotations 

21 

22import dataclasses 

23import re 

24from typing import TYPE_CHECKING, Any, cast 

25 

26from lexigram.contracts.exceptions.domain import ( 

27 AuthenticationError, 

28 AuthorizationError, 

29 ConflictError, 

30 DomainError, 

31 NotFoundError, 

32 PermissionDeniedError, 

33 RateLimitError, 

34 ValidationError, 

35) 

36 

37if TYPE_CHECKING: 

38 from starlette.responses import Response 

39 

40# Registry: error type → HTTP status code 

41_ERROR_STATUS_REGISTRY: list[tuple[type[Exception], int]] = [ 

42 (NotFoundError, 404), 

43 (ValidationError, 422), 

44 (PermissionDeniedError, 403), 

45 (AuthorizationError, 403), 

46 (AuthenticationError, 401), 

47 (ConflictError, 409), 

48 (RateLimitError, 429), 

49 (DomainError, 400), 

50] 

51 

52 

53def _serialize_details(value: Any) -> Any: 

54 """Recursively convert dataclass instances to plain dicts for JSON safety.""" 

55 import dataclasses 

56 

57 if dataclasses.is_dataclass(value) and not isinstance(value, type): 

58 return dataclasses.asdict(value) 

59 if isinstance(value, dict): 

60 return {k: _serialize_details(v) for k, v in value.items()} 

61 if isinstance(value, list): 

62 return [_serialize_details(item) for item in value] 

63 return value 

64 

65 

66def _get_status(error: Exception) -> int: 

67 """Return the HTTP status code for an error, using MRO-aware lookup. 

68 

69 Checks: 

70 1. HTTPError.status_code attribute (web framework errors) 

71 2. Registry mapping by exception type 

72 3. ``DomainError`` subclasses → 400 (client fault); 

73 anything else → 500 (server fault) 

74 """ 

75 # Check for HTTPError.status_code attribute first (web framework errors) 

76 if hasattr(error, "status_code"): 

77 return cast("int", error.status_code) 

78 

79 # Check registry for domain errors (contracts exceptions) 

80 for error_type, status in _ERROR_STATUS_REGISTRY: 

81 if isinstance(error, error_type): 

82 return status 

83 

84 # Domain errors not in the registry are client faults; everything else 

85 # is an unexpected server-side failure and must not read as 4xx. 

86 return 400 if isinstance(error, DomainError) else 500 

87 

88 

89def _exception_type_urn(exc: Exception) -> str: 

90 """Derive a urn:lexigram:{slug} type URI from an exception class name.""" 

91 name = type(exc).__name__ 

92 for suffix in ("Error", "Exception"): 

93 if name.endswith(suffix): 

94 name = name[: -len(suffix)] 

95 break 

96 slug = re.sub(r"(?<!^)(?=[A-Z])", "-", name).lower() 

97 return f"urn:lexigram:{slug}" 

98 

99 

100class ResultResponseMapper: 

101 """Maps ``Result`` error values to HTTP responses. 

102 

103 The default mappings follow REST conventions: 

104 

105 | Error type | Status | 

106 |--------------------------|--------| 

107 | ``NotFoundError`` | 404 | 

108 | ``ValidationError`` | 422 | 

109 | ``PermissionDeniedError``| 403 | 

110 | ``AuthorizationError`` | 403 | 

111 | ``AuthenticationError`` | 401 | 

112 | ``ConflictError`` | 409 | 

113 | ``RateLimitError`` | 429 | 

114 | ``DomainError`` (base) | 400 | 

115 | other ``DomainError`` | 400 (client fault) | 

116 | non-domain error | 500 (server fault) | 

117 

118 Use :meth:`register` to override the status for a specific error type. 

119 """ 

120 

121 def to_response(self, result: Any, success_status: int = 200) -> Response: 

122 """Convert a Result to an HTTP response. 

123 

124 Args: 

125 result: The ``Result[T, E]`` object to map. 

126 success_status: HTTP status code for Ok results (default 200). 

127 

128 Returns: 

129 A :class:`~lexigram.web.transport.responses.JSONResponse` with 

130 appropriate status code and structured body. 

131 """ 

132 

133 from lexigram.web.transport.responses import JSONResponse 

134 

135 if hasattr(result, "is_ok") and callable(result.is_ok): 

136 if result.is_ok(): 

137 return JSONResponse(content=result.unwrap(), status_code=success_status) 

138 return self.error_to_response(result.unwrap_err()) 

139 

140 return JSONResponse(content={"error": "Invalid result object"}, status_code=400) 

141 

142 @classmethod 

143 def register(cls, error_type: type[Exception], status_code: int) -> None: 

144 """Register a custom error type → HTTP status mapping. 

145 

146 Custom registrations take precedence over defaults (inserted at front). 

147 

148 Args: 

149 error_type: The exception class to match. 

150 status_code: HTTP status code to return. 

151 """ 

152 _ERROR_STATUS_REGISTRY.insert(0, (error_type, status_code)) 

153 

154 @classmethod 

155 def error_to_response(cls, error: Any) -> Response: 

156 """Convert a domain error to a JSON HTTP response. 

157 

158 Args: 

159 error: The error value from ``Result.unwrap_err()``. 

160 

161 Returns: 

162 A :class:`~lexigram.web.transport.responses.JSONResponse` with the 

163 appropriate status code and a structured body. 

164 """ 

165 

166 from lexigram.web.errors.problem_detail import ProblemDetail 

167 from lexigram.web.transport.responses import JSONResponse 

168 

169 status = _get_status(error) if isinstance(error, Exception) else 400 

170 

171 if isinstance(error, Exception): 

172 pd = ProblemDetail.from_exception( 

173 error, status=status, type=_exception_type_urn(error) 

174 ) 

175 # Populate errors[] from ValidationError.errors (FieldError list) 

176 field_errors = getattr(error, "errors", None) 

177 if isinstance(field_errors, list) and field_errors: 

178 serialized = [] 

179 for fe in field_errors: 

180 if hasattr(fe, "field") and hasattr(fe, "message"): 

181 serialized.append( 

182 { 

183 "field": fe.field, 

184 "message": fe.message, 

185 "code": getattr(fe, "code", None), 

186 } 

187 ) 

188 elif isinstance(fe, dict): 

189 serialized.append(fe) 

190 if serialized: 

191 pd = dataclasses.replace(pd, errors=serialized) 

192 else: 

193 pd = ProblemDetail(status=status, detail=str(error)) 

194 

195 return JSONResponse( 

196 content=pd.to_dict(), 

197 status_code=status, 

198 media_type="application/problem+json", 

199 ) 

200 

201 

202def error_status(error_type: type[Exception], status_code: int) -> Any: 

203 """Decorator to register a custom error → HTTP status mapping. 

204 

205 Args: 

206 error_type: Exception class to match. 

207 status_code: HTTP status code to use. 

208 

209 Example:: 

210 

211 @error_status(PaymentDeclined, 402) 

212 class BillingController(Controller): 

213 ... 

214 """ 

215 

216 def decorator(cls_or_fn: Any) -> Any: 

217 ResultResponseMapper.register(error_type, status_code) 

218 return cls_or_fn 

219 

220 return decorator 

221 

222 

223__all__ = ["ResultResponseMapper", "error_status"]