API_INTEGRATION.PY

class AdvancedAPIClient(ResonanceOSAPIClient): """Advanced API client with enterprise features""" def __init__(self, base_url: str, api_key: str = None, retry_config: dict = None, cache_config: dict = None): super().__init__(base_url) # Authentication setup if api_key: self.session.headers['Authorization'] = f'Bearer {api_key}' # Retry configuration self.retry_config = retry_config or { "max_retries": 3, "backoff_factor": 2, "retry_on_status": [500, 502, 503, 504] } # Cache configuration self.cache_config = cache_config or { "enabled": True, "ttl": 300, # 5 minutes "max_size": 1000 } # Initialize cache self._cache = {} if self.cache_config["enabled"] else None # Rate limiting self.rate_limiter = RateLimiter( max_requests=100, time_window=60 # 100 requests per minute ) def generate_content_with_retry(self, prompt: str, **kwargs) -> Dict[str, Any]: """Generate content with automatic retry logic""" # Check cache first cache_key = self._generate_cache_key("generate", prompt, kwargs) if self._cache and cache_key in self._cache: cached_result = self._cache[cache_key] if time.time() - cached_result["timestamp"] < self.cache_config["ttl"]: return cached_result["data"] # Rate limiting check self.rate_limiter.wait_if_needed() # Implement retry logic last_exception = None for attempt in range(self.retry_config["max_retries"] + 1): try: result = super().generate_content(prompt, **kwargs) # Cache successful result if self._cache and result.get("status") != "error": self._cache[cache_key] = { "data": result, "timestamp": time.time() } self._cleanup_cache() return result except requests.exceptions.RequestException as e: last_exception = e # Check if we should retry if attempt < self.retry_config["max_retries"]: if (hasattr(e, 'response') and e.response.status_code in self.retry_config["retry_on_status"]): # Exponential backoff backoff_time = self.retry_config["backoff_factor"] ** attempt time.sleep(backoff_time) continue break # Return error if all retries failed return { "status": "error", "message": f"Max retries exceeded: {str(last_exception)}", "attempts": attempt + 1 } def batch_generate(self, prompts: List[str], **kwargs) -> List[Dict[str, Any]]: """Generate content for multiple prompts efficiently""" results = [] # Use threading for concurrent requests with ThreadPoolExecutor(max_workers=5) as executor: # Submit all requests future_to_prompt = { executor.submit(self.generate_content_with_retry, prompt, **kwargs): prompt for prompt in prompts } # Collect results as they complete for future in as_completed(future_to_prompt): prompt = future_to_prompt[future] try: result = future.result() results.append({ "prompt": prompt, "result": result }) except Exception as e: results.append({ "prompt": prompt, "result": { "status": "error", "message": str(e) } }) return results def _generate_cache_key(self, operation: str, prompt: str, kwargs: dict) -> str: """Generate cache key for requests""" import hashlib key_data = f"{operation}:{prompt}:{json.dumps(kwargs, sort_keys=True)}" return hashlib.md5(key_data.encode()).hexdigest() def _cleanup_cache(self): """Remove expired cache entries""" if not self._cache: return current_time = time.time() expired_keys = [ key for key, value in self._cache.items() if current_time - value["timestamp"] > self.cache_config["ttl"] ] for key in expired_keys: del self._cache[key] # Remove oldest entries if cache is too large if len(self._cache) > self.cache_config["max_size"]: sorted_items = sorted( self._cache.items(), key=lambda x: x[1]["timestamp"] ) # Remove oldest entries excess_count = len(self._cache) - self.cache_config["max_size"] for key, _ in sorted_items[:excess_count]: del self._cache[key]