ailine/backend/app/model_services/base.py

"""
模型服务获取器基类和自动降级机制模块

本模块提供：
1. 统一的服务获取器基类，支持服务可用性检查和自动降级
2. 单例模式的服务管理器，确保全局只有一个服务实例
3. 支持链式降级策略，主服务失败时自动尝试备用服务

主要功能：
- BaseServiceProvider：所有服务获取器的基类
- FallbackServiceChain：链式降级处理器
- SingletonServiceManager：单例服务管理器
"""

import abc
from typing import Generic, TypeVar, List, Optional, Any, Callable
from functools import wraps
import logging

logger = logging.getLogger(__name__)

T = TypeVar('T')


class BaseServiceProvider(abc.ABC, Generic[T]):
    """
    服务获取器基类，所有具体服务获取器都需要继承此类
    """
    
    def __init__(self, name: str):
        self._name = name
        self._service_instance: Optional[T] = None
    
    @abc.abstractmethod
    def is_available(self) -> bool:
        """
        检查服务是否可用
        
        Returns:
            bool: 服务是否可用
        """
        pass
    
    @abc.abstractmethod
    def get_service(self) -> T:
        """
        获取服务实例
        
        Returns:
            T: 服务实例
        """
        pass
    
    @property
    def name(self) -> str:
        """获取服务名称"""
        return self._name


class FallbackServiceChain(Generic[T]):
    """
    链式降级处理器，支持多级备用服务
    """
    
    def __init__(self, primary: BaseServiceProvider[T], fallbacks: List[BaseServiceProvider[T]]):
        self._primary = primary
        self._fallbacks = fallbacks
        self._providers = [primary] + fallbacks
    
    def get_available_service(self) -> T:
        """
        获取第一个可用的服务
        
        Returns:
            T: 可用的服务实例
            
        Raises:
            RuntimeError: 如果没有可用的服务
        """
        for provider in self._providers:
            try:
                if provider.is_available():
                    logger.info(f"使用服务: {provider.name}")
                    return provider.get_service()
                else:
                    logger.warning(f"服务不可用: {provider.name}，尝试下一个...")
            except Exception as e:
                logger.warning(f"服务 {provider.name} 检查失败: {e}，尝试下一个...")
        
        raise RuntimeError(f"没有可用的服务，尝试了: {[p.name for p in self._providers]}")
    
    def get_all_providers(self) -> List[BaseServiceProvider[T]]:
        """
        获取所有服务提供者（主服务 + 备用服务）
        
        Returns:
            List[BaseServiceProvider[T]]: 服务提供者列表
        """
        return self._providers.copy()


class SingletonServiceManager:
    """
    单例服务管理器，确保全局只有一个服务实例
    """
    
    _instances: dict = {}
    
    @classmethod
    def get_or_create(cls, key: str, creator: Callable[[], Any]) -> Any:
        """
        获取或创建单例实例
        
        Args:
            key: 单例键
            creator: 创建函数
            
        Returns:
            Any: 单例实例
        """
        if key not in cls._instances:
            cls._instances[key] = creator()
            logger.debug(f"创建单例实例: {key}")
        return cls._instances[key]
    
    @classmethod
    def clear(cls, key: Optional[str] = None):
        """
        清除单例实例
        
        Args:
            key: 单例键，如果为 None 则清除所有
        """
        if key is None:
            cls._instances.clear()
            logger.debug("清除所有单例实例")
        elif key in cls._instances:
            del cls._instances[key]
            logger.debug(f"清除单例实例: {key}")