-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy patherrors.py
More file actions
99 lines (75 loc) · 2.81 KB
/
errors.py
File metadata and controls
99 lines (75 loc) · 2.81 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
from typing import Any
import structlog
from fastapi import Request, status
from fastapi.responses import JSONResponse
logger = structlog.get_logger()
class AppError(Exception):
"""Base exception for application-level errors."""
def __init__(
self,
message: str,
code: str = "INTERNAL_ERROR",
status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
detail: Any = None,
) -> None:
self.message = message
self.code = code
self.status_code = status_code
self.detail = detail
super().__init__(message)
class NotFoundError(AppError):
"""Raised when a requested resource does not exist."""
def __init__(self, message: str, detail: Any = None) -> None:
super().__init__(
message=message,
code="NOT_FOUND",
status_code=status.HTTP_404_NOT_FOUND,
detail=detail,
)
class ValidationError(AppError):
"""Raised when input data fails validation."""
def __init__(self, message: str, detail: Any = None) -> None:
super().__init__(
message=message,
code="VALIDATION_ERROR",
status_code=status.HTTP_400_BAD_REQUEST,
detail=detail,
)
class AuthenticationError(AppError):
"""Raised when authentication fails."""
def __init__(self, message: str, detail: Any = None) -> None:
super().__init__(
message=message,
code="AUTHENTICATION_ERROR",
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail,
)
class TokenError(AuthenticationError):
"""Base exception for token-specific errors."""
class InvalidTokenTypeError(TokenError):
"""Raised when a token type (e.g. access vs refresh) does not match expected."""
class TokenDecodeError(TokenError):
"""Raised when a token cannot be decoded (expired, invalid signature, etc.)."""
async def app_error_handler(_request: Request, exc: AppError) -> JSONResponse:
"""Handle AppError exceptions and return structured JSON responses."""
logger.error(
"Application error",
code=exc.code,
message=exc.message,
status_code=exc.status_code,
detail=exc.detail,
)
content = {"error": exc.code, "detail": exc.message}
if exc.detail is not None:
content["context"] = exc.detail
return JSONResponse(
status_code=exc.status_code,
content=content,
)
async def global_exception_handler(_request: Request, exc: Exception) -> JSONResponse:
"""Catch-all handler for unhandled exceptions."""
logger.error("Unhandled exception", error=str(exc), exc_info=True)
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"error": "INTERNAL_ERROR", "detail": "An unexpected error occurred."},
)