exceptions.py
4.22 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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
"""
自定义异常定义
所有业务异常都应该继承自 APIException,
由全局异常处理器统一处理并返回标准格式的错误响应
"""
from fastapi import HTTPException, status
from typing import Optional, Any
class APIException(HTTPException):
"""
API基础异常
所有业务异常的基类,可以被全局异常处理器捕获和统一处理
"""
def __init__(
self,
status_code: int = status.HTTP_400_BAD_REQUEST,
detail: str = None,
error_code: str = None,
data: Any = None,
headers: dict = None,
):
super().__init__(status_code=status_code, detail=detail, headers=headers)
self.error_code = error_code or "UNKNOWN_ERROR"
self.data = data
class UnauthorizedException(APIException):
"""未授权异常 - 认证失败"""
def __init__(self, detail: str = "未授权", error_code: str = "UNAUTHORIZED"):
super().__init__(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail,
error_code=error_code
)
class ForbiddenException(APIException):
"""禁止访问异常 - 权限不足"""
def __init__(self, detail: str = "禁止访问", error_code: str = "FORBIDDEN"):
super().__init__(
status_code=status.HTTP_403_FORBIDDEN,
detail=detail,
error_code=error_code
)
class NotFoundException(APIException):
"""资源不存在异常"""
def __init__(self, detail: str = "资源不存在", error_code: str = "NOT_FOUND"):
super().__init__(
status_code=status.HTTP_404_NOT_FOUND,
detail=detail,
error_code=error_code
)
class ConflictException(APIException):
"""冲突异常 - 资源已存在"""
def __init__(self, detail: str = "资源已存在", error_code: str = "CONFLICT"):
super().__init__(
status_code=status.HTTP_409_CONFLICT,
detail=detail,
error_code=error_code
)
class ValidationException(APIException):
"""验证异常 - 输入验证失败"""
def __init__(self, detail: str = "验证失败", error_code: str = "VALIDATION_ERROR"):
super().__init__(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=detail,
error_code=error_code
)
class BusinessException(APIException):
"""业务异常 - 业务规则验证失败"""
def __init__(
self,
detail: str = "业务操作失败",
error_code: str = "BUSINESS_ERROR",
status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
):
super().__init__(
status_code=status_code,
detail=detail,
error_code=error_code
)
class InternalServerException(APIException):
"""内部服务器异常"""
def __init__(
self,
detail: str = "内部服务器错误",
error_code: str = "INTERNAL_SERVER_ERROR",
):
super().__init__(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=detail,
error_code=error_code
)
class DatabaseException(APIException):
"""数据库异常"""
def __init__(
self,
detail: str = "数据库操作失败",
error_code: str = "DATABASE_ERROR",
):
super().__init__(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=detail,
error_code=error_code
)
class ExternalServiceException(APIException):
"""外部服务异常 - 调用第三方服务失败"""
def __init__(
self,
detail: str = "外部服务调用失败",
error_code: str = "EXTERNAL_SERVICE_ERROR",
):
super().__init__(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=detail,
error_code=error_code
)
class RateLimitException(APIException):
"""限流异常 - 请求过于频繁"""
def __init__(
self,
detail: str = "请求过于频繁,请稍后再试",
error_code: str = "RATE_LIMIT_EXCEEDED",
):
super().__init__(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=detail,
error_code=error_code
)