356 Zeilen
10 KiB
Python
356 Zeilen
10 KiB
Python
import uuid
|
|
from typing import Optional, Dict, Any
|
|
from datetime import datetime
|
|
|
|
|
|
class BaseApplicationException(Exception):
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
code: str,
|
|
status_code: int = 500,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
user_message: Optional[str] = None
|
|
):
|
|
super().__init__(message)
|
|
self.message = message
|
|
self.code = code
|
|
self.status_code = status_code
|
|
self.details = details or {}
|
|
self.user_message = user_message or message
|
|
self.timestamp = datetime.utcnow()
|
|
self.request_id = str(uuid.uuid4())
|
|
|
|
def to_dict(self, include_details: bool = False) -> Dict[str, Any]:
|
|
result = {
|
|
'error': {
|
|
'code': self.code,
|
|
'message': self.user_message,
|
|
'timestamp': self.timestamp.isoformat(),
|
|
'request_id': self.request_id
|
|
}
|
|
}
|
|
|
|
if include_details and self.details:
|
|
result['error']['details'] = self.details
|
|
|
|
return result
|
|
|
|
|
|
class ValidationException(BaseApplicationException):
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
field: Optional[str] = None,
|
|
value: Any = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
user_message: Optional[str] = None
|
|
):
|
|
details = details or {}
|
|
if field:
|
|
details['field'] = field
|
|
if value is not None:
|
|
details['value'] = str(value)
|
|
|
|
super().__init__(
|
|
message=message,
|
|
code='VALIDATION_ERROR',
|
|
status_code=400,
|
|
details=details,
|
|
user_message=user_message or "Ungültige Eingabe"
|
|
)
|
|
|
|
|
|
class InputValidationError(ValidationException):
|
|
def __init__(
|
|
self,
|
|
field: str,
|
|
message: str,
|
|
value: Any = None,
|
|
expected_type: Optional[str] = None
|
|
):
|
|
details = {'expected_type': expected_type} if expected_type else None
|
|
super().__init__(
|
|
message=f"Invalid input for field '{field}': {message}",
|
|
field=field,
|
|
value=value,
|
|
details=details,
|
|
user_message=f"Ungültiger Wert für Feld '{field}'"
|
|
)
|
|
|
|
|
|
class BusinessRuleViolation(ValidationException):
|
|
def __init__(
|
|
self,
|
|
rule: str,
|
|
message: str,
|
|
context: Optional[Dict[str, Any]] = None
|
|
):
|
|
super().__init__(
|
|
message=message,
|
|
details={'rule': rule, 'context': context or {}},
|
|
user_message="Geschäftsregel verletzt"
|
|
)
|
|
|
|
|
|
class DataIntegrityError(ValidationException):
|
|
def __init__(
|
|
self,
|
|
entity: str,
|
|
constraint: str,
|
|
message: str,
|
|
details: Optional[Dict[str, Any]] = None
|
|
):
|
|
details = details or {}
|
|
details.update({'entity': entity, 'constraint': constraint})
|
|
super().__init__(
|
|
message=message,
|
|
details=details,
|
|
user_message="Datenintegritätsfehler"
|
|
)
|
|
|
|
|
|
class AuthenticationException(BaseApplicationException):
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
user_message: Optional[str] = None
|
|
):
|
|
super().__init__(
|
|
message=message,
|
|
code='AUTHENTICATION_ERROR',
|
|
status_code=401,
|
|
details=details,
|
|
user_message=user_message or "Authentifizierung fehlgeschlagen"
|
|
)
|
|
|
|
|
|
class InvalidCredentialsError(AuthenticationException):
|
|
def __init__(self, username: Optional[str] = None):
|
|
details = {'username': username} if username else None
|
|
super().__init__(
|
|
message="Invalid username or password",
|
|
details=details,
|
|
user_message="Ungültiger Benutzername oder Passwort"
|
|
)
|
|
|
|
|
|
class SessionExpiredError(AuthenticationException):
|
|
def __init__(self, session_id: Optional[str] = None):
|
|
details = {'session_id': session_id} if session_id else None
|
|
super().__init__(
|
|
message="Session has expired",
|
|
details=details,
|
|
user_message="Ihre Sitzung ist abgelaufen"
|
|
)
|
|
|
|
|
|
class InsufficientPermissionsError(AuthenticationException):
|
|
def __init__(
|
|
self,
|
|
required_permission: str,
|
|
user_permissions: Optional[list] = None
|
|
):
|
|
super().__init__(
|
|
message=f"User lacks required permission: {required_permission}",
|
|
details={
|
|
'required': required_permission,
|
|
'user_permissions': user_permissions or []
|
|
},
|
|
user_message="Unzureichende Berechtigungen für diese Aktion"
|
|
)
|
|
self.status_code = 403
|
|
|
|
|
|
class DatabaseException(BaseApplicationException):
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
query: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
user_message: Optional[str] = None
|
|
):
|
|
details = details or {}
|
|
if query:
|
|
details['query_hash'] = str(hash(query))
|
|
|
|
super().__init__(
|
|
message=message,
|
|
code='DATABASE_ERROR',
|
|
status_code=500,
|
|
details=details,
|
|
user_message=user_message or "Datenbankfehler aufgetreten"
|
|
)
|
|
|
|
|
|
class ConnectionError(DatabaseException):
|
|
def __init__(self, message: str, host: Optional[str] = None):
|
|
details = {'host': host} if host else None
|
|
super().__init__(
|
|
message=message,
|
|
details=details,
|
|
user_message="Datenbankverbindung fehlgeschlagen"
|
|
)
|
|
|
|
|
|
class QueryError(DatabaseException):
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
query: str,
|
|
error_code: Optional[str] = None
|
|
):
|
|
super().__init__(
|
|
message=message,
|
|
query=query,
|
|
details={'error_code': error_code} if error_code else None,
|
|
user_message="Datenbankabfrage fehlgeschlagen"
|
|
)
|
|
|
|
|
|
class TransactionError(DatabaseException):
|
|
def __init__(self, message: str, operation: str):
|
|
super().__init__(
|
|
message=message,
|
|
details={'operation': operation},
|
|
user_message="Transaktion fehlgeschlagen"
|
|
)
|
|
|
|
|
|
class ExternalServiceException(BaseApplicationException):
|
|
def __init__(
|
|
self,
|
|
service_name: str,
|
|
message: str,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
user_message: Optional[str] = None
|
|
):
|
|
details = details or {}
|
|
details['service'] = service_name
|
|
|
|
super().__init__(
|
|
message=message,
|
|
code='EXTERNAL_SERVICE_ERROR',
|
|
status_code=502,
|
|
details=details,
|
|
user_message=user_message or f"Fehler beim Zugriff auf {service_name}"
|
|
)
|
|
|
|
|
|
class APIError(ExternalServiceException):
|
|
def __init__(
|
|
self,
|
|
service_name: str,
|
|
endpoint: str,
|
|
status_code: int,
|
|
message: str
|
|
):
|
|
super().__init__(
|
|
service_name=service_name,
|
|
message=message,
|
|
details={
|
|
'endpoint': endpoint,
|
|
'response_status': status_code
|
|
},
|
|
user_message=f"API-Fehler bei {service_name}"
|
|
)
|
|
|
|
|
|
class TimeoutError(ExternalServiceException):
|
|
def __init__(
|
|
self,
|
|
service_name: str,
|
|
timeout_seconds: int,
|
|
operation: str
|
|
):
|
|
super().__init__(
|
|
service_name=service_name,
|
|
message=f"Timeout after {timeout_seconds}s while {operation}",
|
|
details={
|
|
'timeout_seconds': timeout_seconds,
|
|
'operation': operation
|
|
},
|
|
user_message=f"Zeitüberschreitung bei {service_name}"
|
|
)
|
|
|
|
|
|
class ResourceException(BaseApplicationException):
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
resource_type: str,
|
|
resource_id: Any = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
user_message: Optional[str] = None
|
|
):
|
|
details = details or {}
|
|
details.update({
|
|
'resource_type': resource_type,
|
|
'resource_id': str(resource_id) if resource_id else None
|
|
})
|
|
|
|
super().__init__(
|
|
message=message,
|
|
code='RESOURCE_ERROR',
|
|
status_code=404,
|
|
details=details,
|
|
user_message=user_message or "Ressourcenfehler"
|
|
)
|
|
|
|
|
|
class ResourceNotFoundError(ResourceException):
|
|
def __init__(
|
|
self,
|
|
resource_type: str,
|
|
resource_id: Any = None,
|
|
search_criteria: Optional[Dict[str, Any]] = None
|
|
):
|
|
details = {'search_criteria': search_criteria} if search_criteria else None
|
|
super().__init__(
|
|
message=f"{resource_type} not found",
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
details=details,
|
|
user_message=f"{resource_type} nicht gefunden"
|
|
)
|
|
self.status_code = 404
|
|
|
|
|
|
class ResourceConflictError(ResourceException):
|
|
def __init__(
|
|
self,
|
|
resource_type: str,
|
|
resource_id: Any,
|
|
conflict_reason: str
|
|
):
|
|
super().__init__(
|
|
message=f"Conflict with {resource_type}: {conflict_reason}",
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
details={'conflict_reason': conflict_reason},
|
|
user_message=f"Konflikt mit {resource_type}"
|
|
)
|
|
self.status_code = 409
|
|
|
|
|
|
class ResourceLimitExceeded(ResourceException):
|
|
def __init__(
|
|
self,
|
|
resource_type: str,
|
|
limit: int,
|
|
current: int,
|
|
requested: Optional[int] = None
|
|
):
|
|
details = {
|
|
'limit': limit,
|
|
'current': current,
|
|
'requested': requested
|
|
}
|
|
super().__init__(
|
|
message=f"{resource_type} limit exceeded: {current}/{limit}",
|
|
resource_type=resource_type,
|
|
details=details,
|
|
user_message=f"Limit für {resource_type} überschritten"
|
|
)
|
|
self.status_code = 429 |