"""Standardized error handling for API and UI responses.""" from __future__ import annotations import logging from dataclasses import dataclass, field from typing import Optional, Dict, Any from xml.etree.ElementTree import Element, SubElement, tostring from flask import Response, jsonify, request, flash, redirect, url_for, g logger = logging.getLogger(__name__) @dataclass class AppError(Exception): """Base application error with multi-format response support.""" code: str message: str status_code: int = 500 details: Optional[Dict[str, Any]] = field(default=None) def __post_init__(self): super().__init__(self.message) def to_xml_response(self) -> Response: """Convert to S3 API XML error response.""" error = Element("Error") SubElement(error, "Code").text = self.code SubElement(error, "Message").text = self.message request_id = getattr(g, 'request_id', None) if g else None SubElement(error, "RequestId").text = request_id or "unknown" xml_bytes = tostring(error, encoding="utf-8") return Response(xml_bytes, status=self.status_code, mimetype="application/xml") def to_json_response(self) -> tuple[Response, int]: """Convert to JSON error response for UI AJAX calls.""" payload: Dict[str, Any] = { "success": False, "error": { "code": self.code, "message": self.message } } if self.details: payload["error"]["details"] = self.details return jsonify(payload), self.status_code def to_flash_message(self) -> str: """Convert to user-friendly flash message.""" return self.message @dataclass class BucketNotFoundError(AppError): """Bucket does not exist.""" code: str = "NoSuchBucket" message: str = "The specified bucket does not exist" status_code: int = 404 @dataclass class BucketAlreadyExistsError(AppError): """Bucket already exists.""" code: str = "BucketAlreadyExists" message: str = "The requested bucket name is not available" status_code: int = 409 @dataclass class BucketNotEmptyError(AppError): """Bucket is not empty.""" code: str = "BucketNotEmpty" message: str = "The bucket you tried to delete is not empty" status_code: int = 409 @dataclass class ObjectNotFoundError(AppError): """Object does not exist.""" code: str = "NoSuchKey" message: str = "The specified key does not exist" status_code: int = 404 @dataclass class InvalidObjectKeyError(AppError): """Invalid object key.""" code: str = "InvalidKey" message: str = "The specified key is not valid" status_code: int = 400 @dataclass class AccessDeniedError(AppError): """Access denied.""" code: str = "AccessDenied" message: str = "Access Denied" status_code: int = 403 @dataclass class InvalidCredentialsError(AppError): """Invalid credentials.""" code: str = "InvalidAccessKeyId" message: str = "The access key ID you provided does not exist" status_code: int = 403 @dataclass class MalformedRequestError(AppError): """Malformed request.""" code: str = "MalformedXML" message: str = "The XML you provided was not well-formed" status_code: int = 400 @dataclass class InvalidArgumentError(AppError): """Invalid argument.""" code: str = "InvalidArgument" message: str = "Invalid argument" status_code: int = 400 @dataclass class EntityTooLargeError(AppError): """Entity too large.""" code: str = "EntityTooLarge" message: str = "Your proposed upload exceeds the maximum allowed size" status_code: int = 413 def handle_app_error(error: AppError) -> Response: """Handle application errors with appropriate response format.""" log_extra = {"error_code": error.code} if error.details: log_extra["details"] = error.details logger.error(f"{error.code}: {error.message}", extra=log_extra) if request.path.startswith('/ui'): wants_json = ( request.is_json or request.headers.get('X-Requested-With') == 'XMLHttpRequest' or 'application/json' in request.accept_mimetypes.values() ) if wants_json: return error.to_json_response() flash(error.to_flash_message(), 'danger') referrer = request.referrer if referrer and request.host in referrer: return redirect(referrer) return redirect(url_for('ui.buckets_overview')) else: return error.to_xml_response() def register_error_handlers(app): """Register error handlers with a Flask app.""" app.register_error_handler(AppError, handle_app_error) for error_class in [ BucketNotFoundError, BucketAlreadyExistsError, BucketNotEmptyError, ObjectNotFoundError, InvalidObjectKeyError, AccessDeniedError, InvalidCredentialsError, MalformedRequestError, InvalidArgumentError, EntityTooLargeError, ]: app.register_error_handler(error_class, handle_app_error)