""" Base service class providing common functionality and patterns for all services """ import logging import traceback from typing import Dict, Any, Optional, TypeVar, Generic, Callable, Union, Type, cast from functools import wraps from sqlalchemy.orm import Session # Define type variables T = TypeVar('T') R = TypeVar('R') class BaseService: """ Base service class that provides common functionality for all services. Features: - Error handling and logging - Database session management - Decorators for common patterns """ @staticmethod def get_logger(name: str) -> logging.Logger: """ Get a logger with the appropriate namespace Args: name: The service name Returns: Configured logger instance """ return logging.getLogger(f"agent_monitoring_server.services.{name}") @staticmethod def handle_errors(func: Callable[..., R]) -> Callable[..., Union[R, Dict[str, Any]]]: """ Decorator for handling errors in service methods Args: func: The function to decorate Returns: Decorated function with error handling """ @wraps(func) def wrapper(*args, **kwargs) -> Union[R, Dict[str, Any]]: try: return func(*args, **kwargs) except Exception as e: logger = BaseService.get_logger(func.__module__.split('.')[-1]) logger.error(f"Error in {func.__name__}: {str(e)}") logger.error(traceback.format_exc()) # Return error dict return { "status": "error", "message": f"Error in {func.__name__}: {str(e)}" } return wrapper @staticmethod def with_db_session(func: Callable[..., R]) -> Callable[..., Union[R, Dict[str, Any]]]: """ Decorator for handling database sessions in service methods Args: func: The function to decorate Returns: Decorated function with session management """ @wraps(func) def wrapper(*args, **kwargs) -> Union[R, Dict[str, Any]]: # Import here to avoid circular imports from backend.database.utils import get_db session = next(get_db()) try: # Replace session parameter if it exists in kwargs if "session" in kwargs: kwargs["session"] = session return func(*args, **kwargs) except Exception as e: session.rollback() logger = BaseService.get_logger(func.__module__.split('.')[-1]) logger.error(f"Database error in {func.__name__}: {str(e)}") logger.error(traceback.format_exc()) # Return error dict return { "status": "error", "message": f"Database error in {func.__name__}: {str(e)}" } finally: session.close() return wrapper