|
|
""" |
|
|
Groq API client manager with singleton pattern |
|
|
""" |
|
|
import os |
|
|
import threading |
|
|
from typing import Optional |
|
|
from groq import Groq |
|
|
from src.utils.logger import logger |
|
|
from src.config.settings import AppConfig |
|
|
|
|
|
|
|
|
class GroqClientManager: |
|
|
""" |
|
|
🔌 SINGLETON GROQ CLIENT MANAGER |
|
|
Thread-safe singleton pattern for API client |
|
|
""" |
|
|
_instance: Optional['GroqClientManager'] = None |
|
|
_lock = threading.Lock() |
|
|
_client: Optional[Groq] = None |
|
|
_initialized = False |
|
|
|
|
|
def __new__(cls): |
|
|
if cls._instance is None: |
|
|
with cls._lock: |
|
|
if cls._instance is None: |
|
|
cls._instance = super().__new__(cls) |
|
|
return cls._instance |
|
|
|
|
|
def __init__(self): |
|
|
if not self._initialized: |
|
|
with self._lock: |
|
|
if not self._initialized: |
|
|
self._initialize_client() |
|
|
self._initialized = True |
|
|
|
|
|
def _initialize_client(self) -> None: |
|
|
""" |
|
|
🔧 INITIALIZE GROQ CLIENT |
|
|
""" |
|
|
api_key = os.getenv('GROQ_API_KEY') |
|
|
|
|
|
if not api_key: |
|
|
logger.error("❌ GROQ_API_KEY not found in environment variables") |
|
|
raise EnvironmentError( |
|
|
"GROQ_API_KEY not set. Please add it to your .env file:\n" |
|
|
"GROQ_API_KEY=your_api_key_here" |
|
|
) |
|
|
|
|
|
try: |
|
|
self._client = Groq( |
|
|
api_key=api_key, |
|
|
timeout=AppConfig.REQUEST_TIMEOUT |
|
|
) |
|
|
logger.info("✅ Groq client initialized successfully") |
|
|
except Exception as e: |
|
|
logger.error(f"❌ Failed to initialize Groq client: {e}") |
|
|
raise |
|
|
|
|
|
@property |
|
|
def client(self) -> Groq: |
|
|
""" |
|
|
✅ GET GROQ CLIENT INSTANCE |
|
|
""" |
|
|
if self._client is None: |
|
|
raise RuntimeError("Groq client not initialized") |
|
|
return self._client |
|
|
|
|
|
def health_check(self) -> bool: |
|
|
""" |
|
|
🏥 HEALTH CHECK |
|
|
""" |
|
|
try: |
|
|
if self._client is None: |
|
|
logger.warning("⚠️ Health check failed: Client not initialized") |
|
|
return False |
|
|
|
|
|
logger.debug("✅ Health check passed") |
|
|
return True |
|
|
except Exception as e: |
|
|
logger.error(f"❌ Health check failed: {e}") |
|
|
return False |
|
|
|
|
|
def reset(self) -> None: |
|
|
""" |
|
|
🔄 RESET CLIENT (FOR TESTING) |
|
|
""" |
|
|
with self._lock: |
|
|
self._client = None |
|
|
self._initialized = False |
|
|
logger.info("🔄 Groq client reset") |
|
|
|
