225 lines
5.4 KiB
Python
225 lines
5.4 KiB
Python
"""
|
|
自定义异常类
|
|
|
|
定义业务层面的异常,便于统一处理和返回合适的 HTTP 响应。
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
|
|
class AppException(Exception):
|
|
"""应用基础异常"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
code: str = "APP_ERROR",
|
|
details: dict[str, Any] | None = None,
|
|
):
|
|
self.message = message
|
|
self.code = code
|
|
self.details = details or {}
|
|
super().__init__(message)
|
|
|
|
|
|
class AuthenticationError(AppException):
|
|
"""认证错误"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "认证失败",
|
|
code: str = "AUTHENTICATION_ERROR",
|
|
details: dict[str, Any] | None = None,
|
|
):
|
|
super().__init__(message, code, details)
|
|
|
|
|
|
class InvalidCredentialsError(AuthenticationError):
|
|
"""无效凭证"""
|
|
|
|
def __init__(self, message: str = "用户名或密码错误"):
|
|
super().__init__(message, "INVALID_CREDENTIALS")
|
|
|
|
|
|
class TokenError(AuthenticationError):
|
|
"""令牌错误"""
|
|
|
|
def __init__(self, message: str = "令牌无效或已过期"):
|
|
super().__init__(message, "TOKEN_ERROR")
|
|
|
|
|
|
class TokenExpiredError(TokenError):
|
|
"""令牌过期"""
|
|
|
|
def __init__(self, message: str = "令牌已过期"):
|
|
super().__init__(message)
|
|
self.code = "TOKEN_EXPIRED"
|
|
|
|
|
|
class AuthorizationError(AppException):
|
|
"""授权错误"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "权限不足",
|
|
code: str = "AUTHORIZATION_ERROR",
|
|
details: dict[str, Any] | None = None,
|
|
):
|
|
super().__init__(message, code, details)
|
|
|
|
|
|
class ValidationError(AppException):
|
|
"""验证错误"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "数据验证失败",
|
|
code: str = "VALIDATION_ERROR",
|
|
details: dict[str, Any] | None = None,
|
|
):
|
|
super().__init__(message, code, details)
|
|
|
|
|
|
class ResourceNotFoundError(AppException):
|
|
"""资源未找到"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "资源不存在",
|
|
resource_type: str = "resource",
|
|
resource_id: Any = None,
|
|
):
|
|
super().__init__(
|
|
message,
|
|
"RESOURCE_NOT_FOUND",
|
|
{"resource_type": resource_type, "resource_id": resource_id},
|
|
)
|
|
|
|
|
|
class ResourceConflictError(AppException):
|
|
"""资源冲突(如重复创建)"""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str = "资源已存在",
|
|
code: str = "RESOURCE_CONFLICT",
|
|
details: dict[str, Any] | None = None,
|
|
):
|
|
super().__init__(message, code, details)
|
|
|
|
|
|
class UserNotFoundError(ResourceNotFoundError):
|
|
"""用户不存在"""
|
|
|
|
def __init__(self, user_id: Any = None):
|
|
super().__init__("用户不存在", "user", user_id)
|
|
|
|
|
|
class UserAlreadyExistsError(ResourceConflictError):
|
|
"""用户已存在"""
|
|
|
|
def __init__(self, field: str = "username"):
|
|
super().__init__(
|
|
f"该{field}已被注册",
|
|
"USER_ALREADY_EXISTS",
|
|
{"field": field},
|
|
)
|
|
|
|
|
|
class UserDisabledError(AuthenticationError):
|
|
"""用户被禁用"""
|
|
|
|
def __init__(self):
|
|
super().__init__("账户已被禁用", "USER_DISABLED")
|
|
|
|
|
|
class PasswordValidationError(ValidationError):
|
|
"""密码验证错误"""
|
|
|
|
def __init__(self, message: str = "密码不符合要求"):
|
|
super().__init__(message, "PASSWORD_VALIDATION_ERROR")
|
|
|
|
|
|
# ============================================================
|
|
# 余额相关异常
|
|
# ============================================================
|
|
|
|
class InsufficientBalanceError(AppException):
|
|
"""余额不足"""
|
|
|
|
def __init__(self, required: int, available: int):
|
|
super().__init__(
|
|
f"余额不足,需要 {required / 1000:.2f},当前可用 {available / 1000:.2f}",
|
|
"INSUFFICIENT_BALANCE",
|
|
{"required_units": required, "available_units": available},
|
|
)
|
|
|
|
|
|
class DuplicateTransactionError(AppException):
|
|
"""重复交易"""
|
|
|
|
def __init__(self, idempotency_key: str):
|
|
super().__init__(
|
|
"该交易已处理",
|
|
"DUPLICATE_TRANSACTION",
|
|
{"idempotency_key": idempotency_key},
|
|
)
|
|
|
|
|
|
class ConcurrencyError(AppException):
|
|
"""并发冲突"""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
"操作冲突,请重试",
|
|
"CONCURRENCY_ERROR",
|
|
)
|
|
|
|
|
|
# ============================================================
|
|
# 兑换码相关异常
|
|
# ============================================================
|
|
|
|
class RedeemCodeNotFoundError(AppException):
|
|
"""兑换码不存在"""
|
|
|
|
def __init__(self, code: str):
|
|
super().__init__(
|
|
"兑换码不存在",
|
|
"REDEEM_CODE_NOT_FOUND",
|
|
{"code": code},
|
|
)
|
|
|
|
|
|
class RedeemCodeInvalidError(AppException):
|
|
"""兑换码无效"""
|
|
|
|
def __init__(self, code: str, reason: str):
|
|
super().__init__(
|
|
f"兑换码无效: {reason}",
|
|
"REDEEM_CODE_INVALID",
|
|
{"code": code, "reason": reason},
|
|
)
|
|
|
|
|
|
class RedeemCodeExpiredError(AppException):
|
|
"""兑换码已过期"""
|
|
|
|
def __init__(self, code: str):
|
|
super().__init__(
|
|
"兑换码已过期",
|
|
"REDEEM_CODE_EXPIRED",
|
|
{"code": code},
|
|
)
|
|
|
|
|
|
class RedeemCodeUsedError(AppException):
|
|
"""兑换码已使用"""
|
|
|
|
def __init__(self, code: str):
|
|
super().__init__(
|
|
"兑换码已使用",
|
|
"REDEEM_CODE_USED",
|
|
{"code": code},
|
|
)
|