# -*- coding: utf-8 -*-
"""Location: ./mcpgateway/utils/error_formatter.py
Copyright 2025
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti
ContextForge Centralized for Pydantic validation error, SQL exception.
This module provides centralized error formatting for ContextForge,
transforming technical Pydantic validation errors and SQLAlchemy database
exceptions into user-friendly messages suitable for API responses.
The ErrorFormatter class handles:
- Pydantic ValidationError formatting
- SQLAlchemy DatabaseError and IntegrityError formatting
- Mapping technical error messages to user-friendly explanations
- Consistent error response structure
Examples:
>>> from mcpgateway.utils.error_formatter import ErrorFormatter
>>> from pydantic import ValidationError
>>>
>>> # Format validation errors
>>> formatter = ErrorFormatter()
>>> # formatted_error = formatter.format_validation_error(validation_error)
"""
# Standard
from typing import Any, Dict
# Third-Party
from pydantic import ValidationError
from sqlalchemy.exc import DatabaseError, IntegrityError
# First-Party
from mcpgateway.services.logging_service import LoggingService
# Initialize logging service first
logging_service = LoggingService()
logger = logging_service.get_logger(__name__)
class ErrorFormatter:
"""Transform technical errors into user-friendly messages.
Provides static methods to convert Pydantic validation errors and
SQLAlchemy database exceptions into consistent, user-friendly error
responses suitable for API consumption.
Examples:
>>> formatter = ErrorFormatter()
>>> isinstance(formatter, ErrorFormatter)
True
"""
@staticmethod
def format_validation_error(error: ValidationError) -> Dict[str, Any]:
"""Convert Pydantic errors to user-friendly format.
Transforms Pydantic ValidationError objects into a structured
dictionary containing user-friendly error messages. Maps technical
validation messages to more understandable explanations.
Args:
error (ValidationError): The Pydantic validation error to format
Returns:
Dict[str, Any]: A dictionary with formatted error details containing:
- message: General error description
- details: List of field-specific errors
- success: Always False for errors
Examples:
>>> from pydantic import BaseModel, ValidationError, field_validator
>>> # Create a test model with validation
>>> class TestModel(BaseModel):
... name: str
... @field_validator('name')
... def validate_name(cls, v):
... if not v.startswith('A'):
... raise ValueError('Tool name must start with a letter, number, or underscore')
... return v
>>> # Test validation error formatting
>>> try:
... TestModel(name="B123")
... except ValidationError as e:
... print(type(e))
... result = ErrorFormatter.format_validation_error(e)
<class 'pydantic_core._pydantic_core.ValidationError'>
>>> result['message']
'Validation failed: Name must start with a letter, number, or underscore and contain only letters, numbers, periods, underscores, hyphens, and slashes'
>>> result['success']
False
>>> len(result['details']) > 0
True
>>> result['details'][0]['field']
'name'
>>> 'must start with a letter, number, or underscore' in result['details'][0]['message']
True
>>> # Test with multiple errors
>>> class MultiFieldModel(BaseModel):
... name: str
... url: str
... @field_validator('name')
... def validate_name(cls, v):
... if len(v) > 255:
... raise ValueError('Tool name exceeds maximum length')
... return v
... @field_validator('url')
... def validate_url(cls, v):
... if not v.startswith('http'):
... raise ValueError('Tool URL must start with http')
... return v
>>>
>>> try:
... MultiFieldModel(name='A' * 300, url='ftp://invalid')
... except ValidationError as e:
... print(type(e))
... result = ErrorFormatter.format_validation_error(e)
<class 'pydantic_core._pydantic_core.ValidationError'>
>>> len(result['details'])
2
>>> any('too long' in detail['message'] for detail in result['details'])
True
>>> any('valid HTTP' in detail['message'] for detail in result['details'])
True
"""
errors = []
for err in error.errors():
loc = err.get("loc", ["field"])
field = loc[-1] if loc else "field"
msg = err.get("msg", "Invalid value")
# Map technical messages to user-friendly ones
user_message = ErrorFormatter._get_user_message(field, msg)
errors.append({"field": field, "message": user_message})
# Log the full error for debugging
logger.debug(f"Validation error: {error}")
return {"message": f"Validation failed: {user_message}", "details": errors, "success": False}
@staticmethod
def _get_user_message(field: str, technical_msg: str) -> str:
"""Map technical validation messages to user-friendly ones.
Converts technical validation error messages into user-friendly
explanations based on pattern matching. Provides field-specific
context in the returned message.
Args:
field (str): The field name that failed validation
technical_msg (str): The technical validation message from Pydantic
Returns:
str: User-friendly error message with field context
Examples:
>>> # Test letter requirement mapping
>>> msg = ErrorFormatter._get_user_message("name", "Tool name must start with a letter, number, or underscore")
>>> msg
'Name must start with a letter, number, or underscore and contain only letters, numbers, periods, underscores, hyphens, and slashes'
>>> # Test length validation mapping
>>> msg = ErrorFormatter._get_user_message("description", "Tool name exceeds maximum length")
>>> msg
'Description is too long (maximum 255 characters)'
>>> # Test URL validation mapping
>>> msg = ErrorFormatter._get_user_message("endpoint", "Tool URL must start with http")
>>> msg
'Endpoint must be a valid HTTP or WebSocket URL'
>>> # Test directory traversal validation
>>> msg = ErrorFormatter._get_user_message("path", "cannot contain directory traversal")
>>> msg
'Path contains invalid characters'
>>> # Test HTML injection validation
>>> msg = ErrorFormatter._get_user_message("content", "contains HTML tags")
>>> msg
'Content cannot contain HTML or script tags'
>>> # Test fallback for unknown messages
>>> msg = ErrorFormatter._get_user_message("custom_field", "Some unknown error")
>>> msg
'Invalid custom_field'
"""
mappings = {
"Tool name must start with a letter, number, or underscore": f"{field.title()} must start with a letter, number, or underscore and contain only letters, numbers, periods, underscores, hyphens, and slashes",
"Tool name exceeds maximum length": f"{field.title()} is too long (maximum 255 characters)",
"Tool URL must start with": f"{field.title()} must be a valid HTTP or WebSocket URL",
"cannot contain directory traversal": f"{field.title()} contains invalid characters",
"contains HTML tags": f"{field.title()} cannot contain HTML or script tags",
"Server ID must be a valid UUID format": f"{field.title()} must be a valid UUID",
}
for pattern, friendly_msg in mappings.items():
if pattern in technical_msg:
return friendly_msg
# Default fallback
return f"Invalid {field}"
@staticmethod
def format_database_error(error: DatabaseError) -> Dict[str, Any]:
"""Convert database errors to user-friendly format.
Transforms SQLAlchemy database exceptions into structured error
responses. Handles common integrity constraint violations and
provides specific messages for known error patterns.
Args:
error (DatabaseError): The SQLAlchemy database error to format
Returns:
Dict[str, Any]: A dictionary with formatted error details containing:
- message: User-friendly error description
- success: Always False for errors
Examples:
>>> from unittest.mock import Mock
>>>
>>> # Test UNIQUE constraint on gateway URL
>>> mock_error = Mock(spec=IntegrityError)
>>> mock_error.orig = Mock()
>>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: gateways.url"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'A gateway with this URL already exists'
>>> result['success']
False
>>> # Test UNIQUE constraint on gateway slug
>>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: gateways.slug"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'A gateway with this name already exists'
>>> # Test UNIQUE constraint on tool name
>>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: tools.name"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'A tool with this name already exists'
>>> # Test UNIQUE constraint on resource URI
>>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: resources.uri"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'A resource with this URI already exists'
>>> # Test UNIQUE constraint on server name
>>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: servers.name"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'A server with this name already exists'
>>> # Test UNIQUE constraint on prompt name
>>> mock_error.orig.__str__ = lambda self: "UNIQUE constraint failed: prompts.name"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'A prompt with this name already exists'
>>> # Test FOREIGN KEY constraint
>>> mock_error.orig.__str__ = lambda self: "FOREIGN KEY constraint failed"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'Referenced item not found'
>>> # Test NOT NULL constraint
>>> mock_error.orig.__str__ = lambda self: "NOT NULL constraint failed"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'Required field is missing'
>>> # Test CHECK constraint
>>> mock_error.orig.__str__ = lambda self: "CHECK constraint failed: invalid_data"
>>> result = ErrorFormatter.format_database_error(mock_error)
>>> result['message']
'Validation failed. Please check the input data.'
>>> # Test generic database error
>>> generic_error = Mock(spec=DatabaseError)
>>> generic_error.orig = None
>>> result = ErrorFormatter.format_database_error(generic_error)
>>> result['message']
'Unable to complete the operation. Please try again.'
>>> result['success']
False
"""
error_str = str(error.orig) if hasattr(error, "orig") else str(error)
# Log full error
logger.error(f"Database error: {error}")
# Map common database errors
if isinstance(error, IntegrityError):
if "UNIQUE constraint failed" in error_str:
if "gateways.url" in error_str:
return {"message": "A gateway with this URL already exists", "success": False}
elif "gateways.slug" in error_str:
return {"message": "A gateway with this name already exists", "success": False}
elif "tools.name" in error_str:
return {"message": "A tool with this name already exists", "success": False}
elif "resources.uri" in error_str:
return {"message": "A resource with this URI already exists", "success": False}
elif "servers.name" in error_str:
return {"message": "A server with this name already exists", "success": False}
elif "prompts.name" in error_str:
return {"message": "A prompt with this name already exists", "success": False}
elif "servers.id" in error_str:
return {"message": "A server with this ID already exists", "success": False}
elif "a2a_agents.slug" in error_str:
return {"message": "An A2A agent with this name already exists", "success": False}
elif "FOREIGN KEY constraint failed" in error_str:
return {"message": "Referenced item not found", "success": False}
elif "NOT NULL constraint failed" in error_str:
return {"message": "Required field is missing", "success": False}
elif "CHECK constraint failed:" in error_str:
return {"message": "Validation failed. Please check the input data.", "success": False}
# Generic database error
return {"message": "Unable to complete the operation. Please try again.", "success": False}
