Coverage for session_buddy / utils / error_handlers.py: 32.81%
52 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-04 00:43 -0800
1#!/usr/bin/env python3
2"""Error handling utilities for MCP tools.
4This module provides reusable error handling patterns to eliminate code duplication
5across tool implementations.
6"""
8from __future__ import annotations
10import typing as t
11from typing import Any, TypeVar
13if t.TYPE_CHECKING:
14 from collections.abc import Awaitable, Callable
16T = TypeVar("T")
19def _get_logger() -> t.Any:
20 """Lazy logger resolution using standard logging."""
21 import logging
23 return logging.getLogger(__name__)
26class ToolError(Exception):
27 """Base exception for tool errors."""
30class DatabaseUnavailableError(ToolError):
31 """Exception raised when database is not available."""
34class ValidationError(ToolError):
35 """Exception raised when input validation fails."""
38async def handle_tool_errors[T](
39 operation: Callable[..., Awaitable[T]],
40 error_prefix: str = "Operation",
41 *args: Any,
42 **kwargs: Any,
43) -> T | str:
44 """Generic error handler for tool operations.
46 This utility wraps async operations with consistent error handling and logging.
47 Eliminates the need for repetitive try/except blocks in tool implementations.
49 Args:
50 operation: Async function to execute
51 error_prefix: Description of the operation for error messages
52 *args: Positional arguments to pass to operation
53 **kwargs: Keyword arguments to pass to operation
55 Returns:
56 Result from operation, or error message string on failure
58 Example:
59 >>> async def my_operation(x: int) -> int:
60 ... return x * 2
61 >>> result = await handle_tool_errors(my_operation, "Multiplication", 5)
62 >>> print(result)
63 10
65 """
66 try:
67 return await operation(*args, **kwargs)
68 except DatabaseUnavailableError as e:
69 # Don't log database unavailable as exception - it's expected
70 return f"❌ {e!s}"
71 except ValidationError as e:
72 # Don't log validation errors as exceptions - they're user errors
73 return f"❌ {error_prefix} validation failed: {e!s}"
74 except Exception as e:
75 _get_logger().exception(f"Error in {error_prefix}: {e}")
76 return f"❌ {error_prefix} failed: {e!s}"
79async def handle_tool_errors_with_result[T](
80 operation: Callable[..., Awaitable[T]],
81 error_prefix: str = "Operation",
82 *args: Any,
83 **kwargs: Any,
84) -> dict[str, Any]:
85 """Generic error handler that returns structured result dictionary.
87 Similar to handle_tool_errors but returns a dictionary with success/error fields
88 instead of a string. Useful for tools that need structured responses.
90 Args:
91 operation: Async function to execute
92 error_prefix: Description of the operation for error messages
93 *args: Positional arguments to pass to operation
94 **kwargs: Keyword arguments to pass to operation
96 Returns:
97 Dictionary with 'success' bool and either 'data' or 'error' field
99 Example:
100 >>> result = await handle_tool_errors_with_result(operation, "Test")
101 >>> if result["success"]:
102 ... print(result["data"])
103 ... else:
104 ... print(result["error"])
106 """
107 try:
108 data = await operation(*args, **kwargs)
109 return {"success": True, "data": data}
110 except DatabaseUnavailableError as e:
111 return {"success": False, "error": str(e)}
112 except ValidationError as e:
113 return {
114 "success": False,
115 "error": f"{error_prefix} validation failed: {e!s}",
116 }
117 except Exception as e:
118 _get_logger().exception(f"Error in {error_prefix}: {e}")
119 return {"success": False, "error": f"{error_prefix} failed: {e!s}"}
122def validate_required(value: Any, field_name: str) -> None:
123 """Validate that a required field is present and non-empty.
125 Args:
126 value: Value to validate
127 field_name: Name of the field for error messages
129 Raises:
130 ValidationError: If value is None, empty string, or empty collection
132 """
133 if value is None: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 msg = f"{field_name} is required"
135 raise ValidationError(msg)
137 if isinstance(value, str) and not value.strip(): 137 ↛ 138line 137 didn't jump to line 138 because the condition on line 137 was never true
138 msg = f"{field_name} cannot be empty"
139 raise ValidationError(msg)
141 if isinstance(value, (list, dict, set, tuple)) and not value: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 msg = f"{field_name} cannot be empty"
143 raise ValidationError(msg)
146def validate_type(value: Any, expected_type: type, field_name: str) -> None:
147 """Validate that a field has the expected type.
149 Args:
150 value: Value to validate
151 expected_type: Expected type for the value
152 field_name: Name of the field for error messages
154 Raises:
155 ValidationError: If value is not of the expected type
157 """
158 if not isinstance(value, expected_type):
159 msg = (
160 f"{field_name} must be {expected_type.__name__}, got {type(value).__name__}"
161 )
162 raise ValidationError(msg)
165def validate_range(
166 value: float, min_val: float, max_val: float, field_name: str
167) -> None:
168 """Validate that a numeric value is within a specified range.
170 Args:
171 value: Value to validate
172 min_val: Minimum allowed value (inclusive)
173 max_val: Maximum allowed value (inclusive)
174 field_name: Name of the field for error messages
176 Raises:
177 ValidationError: If value is outside the specified range
179 """
180 if not isinstance(value, (int, float)):
181 msg = f"{field_name} must be a number"
182 raise ValidationError(msg)
184 if value < min_val or value > max_val:
185 msg = f"{field_name} must be between {min_val} and {max_val}, got {value}"
186 raise ValidationError(msg)