knowledge_search/src/api/base.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
API客户端基类
"""

from abc import ABC, abstractmethod
from typing import Any, Optional
import logging
import time
from functools import wraps


class BaseAPIClient(ABC):
    """API客户端基类，提供统一的重试、缓存、日志功能"""

    def __init__(
        self,
        api_name: str,
        max_retries: int = 3,
        retry_delay: float = 2.0,
        logger: Optional[logging.Logger] = None
    ):
        self.api_name = api_name
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self.logger = logger or logging.getLogger(api_name)

    def with_retry(self, func):
        """重试装饰器"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, self.max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    self.logger.warning(
                        f"[{self.api_name}] Attempt {attempt}/{self.max_retries} failed: {e}"
                    )
                    if attempt < self.max_retries:
                        time.sleep(self.retry_delay)
                    else:
                        self.logger.error(f"[{self.api_name}] All retries failed: {e}")
                        raise
            return None
        return wrapper

    @abstractmethod
    def call(self, *args, **kwargs) -> Any:
        """具体的API调用逻辑，子类实现"""
        pass