Coverage for src/lexigram/web/exceptions.py: 66%

1"""Web framework error classes. 

2 

3Aligned with lexigram-contracts exception hierarchy. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any 

9 

10from lexigram.contracts.exceptions import LexigramError 

11 

12 

13class HTTPError(LexigramError): 

14 """Base HTTP error, compatible with LexigramError.""" 

15 

16 _code: str = "LEX_ERR_WEB_014" 

17 

18 def __init__( 

19 self, 

20 status_code: int, 

21 detail: str = "", 

22 headers: dict[str, str] | None = None, 

23 code: str | None = None, 

24 cause: Exception | None = None, 

25 ) -> None: 

26 self.status_code = status_code 

27 self.detail = detail 

28 self.headers = headers or {} 

29 self.__cause__ = cause 

30 

31 super().__init__( 

32 message=detail, 

33 details={"status_code": status_code, "headers": self.headers}, 

34 cause=cause, 

35 ) 

36 self.code = code or f"HTTP_{status_code}" 

37 

38 

39class NotFoundError(HTTPError): 

40 """404 Not Found.""" 

41 

42 _code: str = "LEX_ERR_WEB_003" 

43 

44 def __init__( 

45 self, 

46 detail: str = "Not Found", 

47 cause: Exception | None = None, 

48 ) -> None: 

49 super().__init__(status_code=404, detail=detail, code="NOT_FOUND", cause=cause) 

50 

51 

52class BadRequestError(HTTPError): 

53 """400 Bad Request.""" 

54 

55 _code: str = "LEX_ERR_WEB_004" 

56 

57 def __init__( 

58 self, 

59 detail: str = "Bad Request", 

60 cause: Exception | None = None, 

61 ) -> None: 

62 super().__init__( 

63 status_code=400, 

64 detail=detail, 

65 code="BAD_REQUEST", 

66 cause=cause, 

67 ) 

68 

69 

70class UnauthorizedError(HTTPError): 

71 """401 Unauthorized.""" 

72 

73 _code: str = "LEX_ERR_WEB_005" 

74 

75 def __init__( 

76 self, 

77 detail: str = "Unauthorized", 

78 cause: Exception | None = None, 

79 ) -> None: 

80 super().__init__( 

81 status_code=401, 

82 detail=detail, 

83 code="UNAUTHORIZED", 

84 cause=cause, 

85 ) 

86 

87 

88class ForbiddenError(HTTPError): 

89 """403 Forbidden.""" 

90 

91 _code: str = "LEX_ERR_WEB_006" 

92 

93 def __init__( 

94 self, 

95 detail: str = "Forbidden", 

96 cause: Exception | None = None, 

97 ) -> None: 

98 super().__init__(status_code=403, detail=detail, code="FORBIDDEN", cause=cause) 

99 

100 

101class MethodNotAllowedError(HTTPError): 

102 """405 Method Not Allowed.""" 

103 

104 _code: str = "LEX_ERR_WEB_007" 

105 

106 def __init__( 

107 self, 

108 detail: str = "Method Not Allowed", 

109 cause: Exception | None = None, 

110 ) -> None: 

111 super().__init__( 

112 status_code=405, 

113 detail=detail, 

114 code="METHOD_NOT_ALLOWED", 

115 cause=cause, 

116 ) 

117 

118 

119class ConflictError(HTTPError): 

120 """409 Conflict.""" 

121 

122 _code: str = "LEX_ERR_WEB_008" 

123 

124 def __init__( 

125 self, 

126 detail: str = "Conflict", 

127 cause: Exception | None = None, 

128 ) -> None: 

129 super().__init__(status_code=409, detail=detail, code="CONFLICT", cause=cause) 

130 

131 

132class UnprocessableEntityError(HTTPError): 

133 """422 Unprocessable Entity.""" 

134 

135 _code: str = "LEX_ERR_WEB_009" 

136 

137 def __init__( 

138 self, 

139 detail: str = "Unprocessable Entity", 

140 cause: Exception | None = None, 

141 ) -> None: 

142 super().__init__( 

143 status_code=422, 

144 detail=detail, 

145 code="UNPROCESSABLE_ENTITY", 

146 cause=cause, 

147 ) 

148 

149 

150class InternalServerError(HTTPError): 

151 """500 Internal Server Error.""" 

152 

153 _code: str = "LEX_ERR_WEB_010" 

154 

155 def __init__( 

156 self, 

157 detail: str = "Internal Server Error", 

158 code: str = "INTERNAL_SERVER_ERROR", 

159 cause: Exception | None = None, 

160 ) -> None: 

161 super().__init__( 

162 status_code=500, 

163 detail=detail, 

164 code=code, 

165 cause=cause, 

166 ) 

167 

168 

169class DependencyResolutionError(InternalServerError): 

170 """500 Dependency Resolution Error.""" 

171 

172 _code: str = "LEX_ERR_WEB_011" 

173 

174 def __init__( 

175 self, 

176 param: str, 

177 service_type: Any, 

178 cause: Exception | None = None, 

179 ) -> None: 

180 self.param = param 

181 self.service_type = service_type 

182 super().__init__( 

183 detail=f"Failed to resolve dependency for parameter '{param}'", 

184 code="dependency_resolution_error", 

185 cause=cause, 

186 ) 

187 self.details.update({"param": param, "service_type": str(service_type)}) 

188 

189 

190class RateLimitError(HTTPError): 

191 """429 Too Many Requests.""" 

192 

193 _code: str = "LEX_ERR_WEB_012" 

194 

195 def __init__( 

196 self, 

197 detail: str = "Too Many Requests", 

198 retry_after: int | None = None, 

199 cause: Exception | None = None, 

200 ) -> None: 

201 headers = {"Retry-After": str(retry_after)} if retry_after else None 

202 super().__init__( 

203 status_code=429, 

204 detail=detail, 

205 headers=headers, 

206 code="RATE_LIMIT_EXCEEDED", 

207 cause=cause, 

208 ) 

209 

210 

211class TooManyConnectionsError(HTTPError): 

212 """503 Service Unavailable — connection limit reached. 

213 

214 Raised when a streaming endpoint (e.g. SSE) has reached its 

215 ``max_connections`` cap and cannot accept further connections. 

216 """ 

217 

218 _code: str = "LEX_ERR_WEB_013" 

219 

220 def __init__( 

221 self, 

222 detail: str = "Too many active connections", 

223 cause: Exception | None = None, 

224 ) -> None: 

225 super().__init__( 

226 status_code=503, 

227 detail=detail, 

228 code="TOO_MANY_CONNECTIONS", 

229 cause=cause, 

230 ) 

231 

232 

233__all__ = [ 

234 "BadRequestError", 

235 "ConflictError", 

236 "DependencyResolutionError", 

237 "ForbiddenError", 

238 "HTTPError", 

239 "InternalServerError", 

240 "MethodNotAllowedError", 

241 "NotFoundError", 

242 "RateLimitError", 

243 "TooManyConnectionsError", 

244 "UnauthorizedError", 

245 "UnprocessableEntityError", 

246]