Spaces:
Sleeping
Sleeping
File size: 3,256 Bytes
c2ea5ed |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
"""
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 |