273 Zeilen
9.6 KiB
Python
273 Zeilen
9.6 KiB
Python
import logging
|
|
import traceback
|
|
from functools import wraps
|
|
from typing import Optional, Dict, Any, Callable, Union
|
|
from flask import (
|
|
Flask, request, jsonify, render_template, flash, redirect,
|
|
url_for, current_app, g
|
|
)
|
|
from werkzeug.exceptions import HTTPException
|
|
import psycopg2
|
|
|
|
from .exceptions import (
|
|
BaseApplicationException, DatabaseException, ValidationException,
|
|
AuthenticationException, ResourceException, QueryError,
|
|
ConnectionError, TransactionError
|
|
)
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def init_error_handlers(app: Flask) -> None:
|
|
@app.before_request
|
|
def before_request():
|
|
g.request_id = request.headers.get('X-Request-ID',
|
|
BaseApplicationException('', '', 0).request_id)
|
|
|
|
@app.errorhandler(BaseApplicationException)
|
|
def handle_application_error(error: BaseApplicationException):
|
|
return _handle_error(error)
|
|
|
|
@app.errorhandler(HTTPException)
|
|
def handle_http_error(error: HTTPException):
|
|
return _handle_error(error)
|
|
|
|
@app.errorhandler(psycopg2.Error)
|
|
def handle_database_error(error: psycopg2.Error):
|
|
db_exception = _convert_psycopg2_error(error)
|
|
return _handle_error(db_exception)
|
|
|
|
@app.errorhandler(Exception)
|
|
def handle_unexpected_error(error: Exception):
|
|
logger.error(
|
|
f"Unexpected error: {str(error)}",
|
|
exc_info=True,
|
|
extra={'request_id': getattr(g, 'request_id', 'unknown')}
|
|
)
|
|
|
|
if current_app.debug:
|
|
raise
|
|
|
|
generic_error = BaseApplicationException(
|
|
message="An unexpected error occurred",
|
|
code="INTERNAL_ERROR",
|
|
status_code=500,
|
|
user_message="Ein unerwarteter Fehler ist aufgetreten"
|
|
)
|
|
return _handle_error(generic_error)
|
|
|
|
|
|
def _handle_error(error: Union[BaseApplicationException, HTTPException, Exception]) -> tuple:
|
|
if isinstance(error, HTTPException):
|
|
status_code = error.code
|
|
error_dict = {
|
|
'error': {
|
|
'code': error.name.upper().replace(' ', '_'),
|
|
'message': error.description or str(error),
|
|
'request_id': getattr(g, 'request_id', 'unknown')
|
|
}
|
|
}
|
|
user_message = error.description or str(error)
|
|
elif isinstance(error, BaseApplicationException):
|
|
status_code = error.status_code
|
|
error_dict = error.to_dict(include_details=current_app.debug)
|
|
error_dict['error']['request_id'] = getattr(g, 'request_id', error.request_id)
|
|
user_message = error.user_message
|
|
|
|
logger.error(
|
|
f"{error.__class__.__name__}: {error.message}",
|
|
extra={
|
|
'error_code': error.code,
|
|
'details': error.details,
|
|
'request_id': error_dict['error']['request_id']
|
|
}
|
|
)
|
|
else:
|
|
status_code = 500
|
|
error_dict = {
|
|
'error': {
|
|
'code': 'INTERNAL_ERROR',
|
|
'message': 'An internal error occurred',
|
|
'request_id': getattr(g, 'request_id', 'unknown')
|
|
}
|
|
}
|
|
user_message = "Ein interner Fehler ist aufgetreten"
|
|
|
|
if _is_json_request():
|
|
return jsonify(error_dict), status_code
|
|
else:
|
|
if status_code == 404:
|
|
return render_template('404.html'), 404
|
|
elif status_code >= 500:
|
|
return render_template('500.html', error=user_message), status_code
|
|
else:
|
|
flash(user_message, 'error')
|
|
return render_template('error.html',
|
|
error=user_message,
|
|
error_code=error_dict['error']['code'],
|
|
request_id=error_dict['error']['request_id']), status_code
|
|
|
|
|
|
def _convert_psycopg2_error(error: psycopg2.Error) -> DatabaseException:
|
|
error_code = getattr(error, 'pgcode', None)
|
|
error_message = str(error).split('\n')[0]
|
|
|
|
if isinstance(error, psycopg2.OperationalError):
|
|
return ConnectionError(
|
|
message=f"Database connection failed: {error_message}",
|
|
host=None
|
|
)
|
|
elif isinstance(error, psycopg2.IntegrityError):
|
|
if error_code == '23505':
|
|
return ValidationException(
|
|
message="Duplicate entry violation",
|
|
details={'constraint': error_message},
|
|
user_message="Dieser Eintrag existiert bereits"
|
|
)
|
|
elif error_code == '23503':
|
|
return ValidationException(
|
|
message="Foreign key violation",
|
|
details={'constraint': error_message},
|
|
user_message="Referenzierte Daten existieren nicht"
|
|
)
|
|
else:
|
|
return ValidationException(
|
|
message="Data integrity violation",
|
|
details={'error_code': error_code},
|
|
user_message="Datenintegritätsfehler"
|
|
)
|
|
elif isinstance(error, psycopg2.DataError):
|
|
return ValidationException(
|
|
message="Invalid data format",
|
|
details={'error': error_message},
|
|
user_message="Ungültiges Datenformat"
|
|
)
|
|
else:
|
|
return QueryError(
|
|
message=error_message,
|
|
query="[query hidden for security]",
|
|
error_code=error_code
|
|
)
|
|
|
|
|
|
def _is_json_request() -> bool:
|
|
return (request.is_json or
|
|
request.path.startswith('/api/') or
|
|
request.accept_mimetypes.best == 'application/json')
|
|
|
|
|
|
def handle_errors(
|
|
catch: tuple = (Exception,),
|
|
message: str = "Operation failed",
|
|
user_message: Optional[str] = None,
|
|
redirect_to: Optional[str] = None
|
|
) -> Callable:
|
|
def decorator(func: Callable) -> Callable:
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except catch as e:
|
|
if isinstance(e, BaseApplicationException):
|
|
raise
|
|
|
|
logger.error(
|
|
f"Error in {func.__name__}: {str(e)}",
|
|
exc_info=True,
|
|
extra={'request_id': getattr(g, 'request_id', 'unknown')}
|
|
)
|
|
|
|
if _is_json_request():
|
|
return jsonify({
|
|
'error': {
|
|
'code': 'OPERATION_FAILED',
|
|
'message': user_message or message,
|
|
'request_id': getattr(g, 'request_id', 'unknown')
|
|
}
|
|
}), 500
|
|
else:
|
|
flash(user_message or message, 'error')
|
|
if redirect_to:
|
|
return redirect(url_for(redirect_to))
|
|
return redirect(request.referrer or url_for('admin.dashboard'))
|
|
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def validate_request(
|
|
required_fields: Optional[Dict[str, type]] = None,
|
|
optional_fields: Optional[Dict[str, type]] = None
|
|
) -> Callable:
|
|
def decorator(func: Callable) -> Callable:
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
data = request.get_json() if request.is_json else request.form
|
|
|
|
if required_fields:
|
|
for field, expected_type in required_fields.items():
|
|
if field not in data:
|
|
raise ValidationException(
|
|
message=f"Missing required field: {field}",
|
|
field=field,
|
|
user_message=f"Pflichtfeld fehlt: {field}"
|
|
)
|
|
|
|
try:
|
|
if expected_type != str:
|
|
if expected_type == int:
|
|
int(data[field])
|
|
elif expected_type == float:
|
|
float(data[field])
|
|
elif expected_type == bool:
|
|
if isinstance(data[field], str):
|
|
if data[field].lower() not in ['true', 'false', '1', '0']:
|
|
raise ValueError
|
|
except (ValueError, TypeError):
|
|
raise ValidationException(
|
|
message=f"Invalid type for field {field}",
|
|
field=field,
|
|
value=data[field],
|
|
details={'expected_type': expected_type.__name__},
|
|
user_message=f"Ungültiger Typ für Feld {field}"
|
|
)
|
|
|
|
return func(*args, **kwargs)
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
class ErrorContext:
|
|
def __init__(
|
|
self,
|
|
operation: str,
|
|
resource_type: Optional[str] = None,
|
|
resource_id: Optional[Any] = None
|
|
):
|
|
self.operation = operation
|
|
self.resource_type = resource_type
|
|
self.resource_id = resource_id
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if exc_val is None:
|
|
return False
|
|
|
|
if isinstance(exc_val, BaseApplicationException):
|
|
return False
|
|
|
|
logger.error(
|
|
f"Error during {self.operation}",
|
|
exc_info=True,
|
|
extra={
|
|
'operation': self.operation,
|
|
'resource_type': self.resource_type,
|
|
'resource_id': self.resource_id,
|
|
'request_id': getattr(g, 'request_id', 'unknown')
|
|
}
|
|
)
|
|
|
|
return False |