"""
Decorators for MCP tools.
Provides error handling, logging, and caching decorators.
"""
from __future__ import annotations
import logging
from functools import wraps
from typing import Any, Callable, ParamSpec, TypeVar
from .exceptions import (
OdooAuthenticationError,
OdooConnectionError,
OdooError,
OdooNotFoundError,
OdooPermissionError,
OdooValidationError,
)
logger = logging.getLogger(__name__)
P = ParamSpec("P")
T = TypeVar("T")
def handle_odoo_errors(func: Callable[P, T]) -> Callable[P, str | T]:
"""
Decorator that catches Odoo exceptions and returns user-friendly error messages.
This decorator should wrap all MCP tool functions to ensure errors are
properly formatted for the user.
Args:
func: The function to wrap
Returns:
Wrapped function that catches exceptions and returns error strings
"""
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> str | T:
try:
return func(*args, **kwargs)
except OdooValidationError as e:
logger.warning(
"Validation error in %s: %s",
func.__name__,
e.message,
extra={"details": e.details},
)
return f"Validation Error: {e.message}"
except OdooNotFoundError as e:
logger.info(
"Record not found in %s: %s",
func.__name__,
e.message,
extra={"model": e.model, "record_id": e.record_id},
)
return f"Not Found: {e.message}"
except OdooAuthenticationError as e:
logger.error(
"Authentication error in %s: %s",
func.__name__,
e.message,
extra={"details": e.details},
)
return f"Authentication Error: {e.message}. Check your credentials."
except OdooConnectionError as e:
logger.error(
"Connection error in %s: %s",
func.__name__,
e.message,
extra={"details": e.details},
)
return f"Connection Error: {e.message}. Check Odoo availability."
except OdooPermissionError as e:
logger.warning(
"Permission denied in %s: %s",
func.__name__,
e.message,
extra={"details": e.details},
)
return f"Permission Denied: {e.message}"
except OdooError as e:
logger.error(
"Odoo error in %s: %s",
func.__name__,
e.message,
extra={"details": e.details},
)
return f"Error: {e.message}"
except Exception as e:
logger.exception(
"Unexpected error in %s: %s",
func.__name__,
str(e),
)
return f"Unexpected Error: {str(e)}"
return wrapper
def log_operation(operation: str) -> Callable[[Callable[P, T]], Callable[P, T]]:
"""
Decorator that logs operation entry and exit.
Args:
operation: Description of the operation being performed
Returns:
Decorator function
"""
def decorator(func: Callable[P, T]) -> Callable[P, T]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
logger.debug(
"Starting operation: %s (%s)",
operation,
func.__name__,
extra={"args_count": len(args), "kwargs_keys": list(kwargs.keys())},
)
try:
result = func(*args, **kwargs)
logger.debug(
"Completed operation: %s (%s)",
operation,
func.__name__,
)
return result
except Exception:
logger.debug(
"Failed operation: %s (%s)",
operation,
func.__name__,
)
raise
return wrapper
return decorator
def retry_on_connection_error(
max_retries: int = 3,
delay: float = 1.0,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
"""
Decorator that retries on connection errors.
Args:
max_retries: Maximum number of retry attempts
delay: Delay between retries in seconds
Returns:
Decorator function
"""
import time
def decorator(func: Callable[P, T]) -> Callable[P, T]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
last_error: Exception | None = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except OdooConnectionError as e:
last_error = e
if attempt < max_retries - 1:
logger.warning(
"Connection failed, retrying in %.1fs (attempt %d/%d)",
delay,
attempt + 1,
max_retries,
)
time.sleep(delay)
# All retries exhausted
raise last_error or OdooConnectionError("Connection failed after retries")
return wrapper
return decorator
