Managing API keys for AI services feels like juggling flaming torches while riding a unicycle. One wrong move and your production system crashes at 3 AM. I've been there. Last year, I watched a startup burn through $4,000 in credits in 48 hours because a single API key got rate-limited during a traffic spike. That's when I built a proper rotation system.
In this guide, I'll walk you through building an AI API key rotation system from absolute scratch. No prior experience required. By the end, you'll have a production-ready solution that handles automatic refreshes and gradual rollouts—keeping your apps online while saving money.
What is API Key Rotation and Why Should You Care?
Think of an API key like a password to a exclusive club. If you use the same password for everything and someone steals it, you're locked out. API key rotation means having multiple keys and cycling through them automatically.
Here's the problem without rotation:
- Your single key hits rate limits during peak traffic
- One expired key crashes your entire application
- Security breaches give attackers full access
- No way to test new AI models without risking production stability
With rotation, you get:
- Zero downtime during key expiration or rate limiting
- Built-in redundancy if one provider goes down
- Safe testing of new AI models via gradual traffic shifting
- Cost optimization by routing requests intelligently
Understanding the HolySheep AI Advantage
Before diving into code, let me explain why signing up for HolySheep AI makes rotation even more valuable. HolySheep offers rates at ¥1 = $1, which saves you 85% compared to typical market rates of ¥7.3 per dollar. They support WeChat and Alipay payments natively, deliver sub-50ms latency, and give you free credits on registration.
The 2026 model pricing structure makes strategic routing essential:
- GPT-4.1: $8 per million tokens output
- Claude Sonnet 4.5: $15 per million tokens output
- Gemini 2.5 Flash: $2.50 per million tokens output
- DeepSeek V3.2: $0.42 per million tokens output
Routing 70% of requests to DeepSeek V3.2 instead of GPT-4.1 could save thousands monthly. But you need rotation to do this safely.
Setting Up Your Environment
First, you'll need Python installed on your computer. Download it from python.org and make sure to check "Add Python to PATH" during installation. Open your terminal (Command Prompt on Windows, Terminal on Mac) and verify installation:
python --version
pip --versionYou should see Python 3.8 or higher. Now create a project folder and install the libraries we need:
mkdir ai-key-rotation
cd ai-key-rotation
pip install requests python-dotenv redis schedule[Screenshot hint: Your terminal should show successful installations with green text and version numbers]
Building Your First Key Rotation System
Step 1: Store Your API Keys Securely
Create a new file called .env in your project folder. This file stores your secrets safely, separate from your code. Never share this file or upload it to GitHub.
# HolySheep AI API Keys (get yours at https://www.holysheep.ai/register)
HOLYSHEEP_KEY_1=hs_live_abc123def456
HOLYSHEEP_KEY_2=hs_live_xyz789ghi012
HOLYSHEEP_KEY_3=hs_live_mno345pqr678
Redis configuration for key storage
REDIS_HOST=localhost
REDIS_PORT=6379
Rotation settings
ROTATION_INTERVAL_MINUTES=60
MAX_REQUESTS_PER_KEY=1000Each HolySheep key starts with hs_live_ for production or hs_test_ for testing. Generate multiple keys from your HolySheep dashboard before continuing.
Step 2: Create the Key Manager Class
Create a file named key_manager.py. This class handles everything about managing your API keys—tracking usage, rotating them, and handling failures gracefully.
import os
import time
import random
from datetime import datetime, timedelta
from collections import deque
from dotenv import load_dotenv
load_dotenv()
class APIKeyManager:
"""
Manages multiple API keys with automatic rotation and health monitoring.
Designed for production systems requiring high availability.
"""
def __init__(self):
self.keys = []
self.key_stats = {}
self.failed_keys = {}
self.current_index = 0
self.last_rotation = datetime.now()
self.rotation_interval = int(os.getenv('ROTATION_INTERVAL_MINUTES', 60))
self.max_requests_per_key = int(os.getenv('MAX_REQUESTS_PER_KEY', 1000))
self._initialize_keys()
def _initialize_keys(self):
"""Load all API keys from environment variables."""
for i in range(1, 10): # Support up to 9 keys
key = os.getenv(f'HOLYSHEEP_KEY_{i}')
if key:
self.keys.append(key)
self.key_stats[key] = {
'requests_made': 0,
'errors': 0,
'last_used': None,
'health_score': 100.0,
'created_at': datetime.now()
}
if not self.keys:
raise ValueError("No API keys found. Please set HOLYSHEEP_KEY_1 in .env")
print(f"✓ Loaded {len(self.keys)} API keys successfully")
def get_available_key(self):
"""
Returns the best available key based on health and usage.
Implements weighted random selection with health bias.
"""
self._check_rotation_needed()
self._remove_failed_keys()
# Calculate weights based on health scores
weights = []
for key in self.keys:
health = self.key_stats[key]['health_score']
weight = max(health, 10) # Minimum weight of 10
weights.append(weight)
# Weighted random selection
selected_key = random.choices(self.keys, weights=weights)[0]
# Update statistics
self.key_stats[selected_key]['requests_made'] += 1
self.key_stats[selected_key]['last_used'] = datetime.now()
return selected_key
def _check_rotation_needed(self):
"""Automatically rotate to fresh keys if interval has passed."""
time_since_rotation = (datetime.now() - self.last_rotation).total_seconds() / 60
if time_since_rotation >= self.rotation_interval:
self.current_index = (self.current_index + 1) % len(self.keys)
self.last_rotation = datetime.now()
print(f"🔄 Rotated to key index {self.current_index}")
def _remove_failed_keys(self):
"""Remove keys that have exceeded error thresholds."""
keys_to_remove = []
for key in self.keys:
stats = self.key_stats[key]
if stats['errors'] >= 10:
keys_to_remove.append(key)
for key in keys_to_remove:
self.keys.remove(key)
print(f"⚠️ Removed failed key: {key[:15]}... (total errors: {self.key_stats[key]['errors']})")
def report_success(self, key):
"""Call this after successful API request."""
self.key_stats[key]['health_score'] = min(100, self.key_stats[key]['health_score'] + 2)
def report_error(self, key, error_type=None):
"""Call this after failed API request to track key health."""
stats = self.key_stats[key]
stats['errors'] += 1
# Decrease health score based on error type
if error_type == 'rate_limit':
stats['health_score'] = max(0, stats['health_score'] - 15)
elif error_type == 'timeout':
stats['health_score'] = max(0, stats['health_score'] - 10)
else:
stats['health_score'] = max(0, stats['health_score'] - 5)
print(f"❌ Key {key[:15]}... reported {error_type}: health now {stats['health_score']:.1f}")
def get_status_report(self):
"""Returns detailed status of all keys."""
report = ["\n📊 API Key Status Report", "=" * 50]
for i, key in enumerate(self.keys):
stats = self.key_stats[key]
status = "🟢 Healthy" if stats['health_score'] > 70 else "🟡 Degraded" if stats['health_score'] > 30 else "🔴 Critical"
report.append(f"\nKey {i+1}: {key[:15]}...")
report.append(f" Status: {status}")
report.append(f" Health Score: {stats['health_score']:.1f}%")
report.append(f" Requests: {stats['requests_made']}")
report.append(f" Errors: {stats['errors']}")
report.append(f" Last Used: {stats['last_used']}")
return "\n".join(report)
Singleton instance for easy importing
key_manager = APIKeyManager()[Screenshot hint: The output shows green checkmarks and key index rotation messages]
Step 3: Build the API Client with Automatic Retry
Create ai_client.py to handle actual API calls with automatic retry logic and error handling. This is where your application meets HolySheep AI.
import requests
import time
from key_manager import key_manager
class HolySheepAIClient:
"""
Production-ready client for HolySheep AI API with automatic key rotation.
Base URL: https://api.holysheep.ai/v1
"""
def __init__(self):
self.base_url = "https://api.holysheep.ai/v1"
self.max_retries = 3
self.timeout = 30
def chat_completion(self, messages, model="gpt-4.1", temperature=0.7, max_tokens=1000):
"""
Send a chat completion request with automatic key rotation.
Args:
messages: List of message dictionaries [{"role": "user", "content": "..."}]
model: AI model to use (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
temperature: Creativity level (0 = deterministic, 1 = creative)
max_tokens: Maximum response length
Returns:
dict: API response or error information
"""
url = f"{self.base_url}/chat/completions"
for attempt in range(self.max_retries):
api_key = key_manager.get_available_key()
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = requests.post(
url,
headers=headers,
json=payload,
timeout=self.timeout
)
if response.status_code == 200:
key_manager.report_success(api_key)
return {"success": True, "data": response.json()}
elif response.status_code == 429:
# Rate limited - rotate key immediately
key_manager.report_error(api_key, "rate_limit")
print(f"⚡ Rate limited, rotating key (attempt {attempt + 1}/{self.max_retries})")
time.sleep(2 ** attempt) # Exponential backoff
elif response.status_code == 401:
# Unauthorized - key might be expired
key_manager.report_error(api_key, "auth_error")
print(f"🔑 Authentication error for key, rotating (attempt {attempt + 1}/{self.max_retries})")
elif response.status_code >= 500:
# Server error - might be temporary
key_manager.report_error(api_key, "server_error")
print(f"🌐 Server error, retrying (attempt {attempt + 1}/{self.max_retries})")
time.sleep(2 ** attempt)
else:
key_manager.report_error(api_key, "unknown_error")
return {
"success": False,
"error": f"HTTP {response.status_code}",
"details": response.text
}
except requests.exceptions.Timeout:
key_manager.report_error(api_key, "timeout")
print(f"⏱️ Request timeout, retrying (attempt {attempt + 1}/{self.max_retries})")
time.sleep(2 ** attempt)
except requests.exceptions.ConnectionError as e:
key_manager.report_error(api_key, "connection_error")
print(f"🔌 Connection error: {e}")
time.sleep(5)
return {
"success": False,
"error": "Max retries exceeded",
"code": "RETRIES_EXHAUSTED"
}
def batch_completion(self, prompts, model="deepseek-v3.2"):
"""
Process multiple prompts efficiently using the same model.
Implements automatic batching with key rotation.
"""
results = []
for i, prompt in enumerate(prompts):
print(f"Processing prompt {i + 1}/{len(prompts)}...")
messages = [{"role": "user", "content": prompt}]
result = self.chat_completion(messages, model=model)
results.append({
"prompt": prompt,
"response": result,
"index": i
})
# Small delay to respect rate limits
time.sleep(0.1)
return results
Example usage
if __name__ == "__main__":
client = HolySheepAIClient()
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain API key rotation in simple terms."}
]
print("\n🚀 Sending request to HolySheep AI...")
result = client.chat_completion(messages, model="deepseek-v3.2")
if result["success"]:
print("✅ Request successful!")
print(f"Response: {result['data']['choices'][0]['message']['content']}")
else:
print(f"❌ Request failed: {result['error']}")
# Print status report
print(key_manager.get_status_report())Step 4: Implementing Canary Release for Model Testing
Now let's add the canary release feature. This lets you gradually shift traffic to new models or new API keys, testing them with a small percentage before full rollout.
import random
from typing import Callable, Any, Dict, List
class CanaryRelease:
"""
Gradual traffic shifting for testing new models or configurations.
Starts with 5% traffic to the new target, scales up based on success.
"""
def __init__(self, key_manager):
self.key_manager = key_manager
self.canary_configs = {}
self.default_traffic_split = {
"deepseek-v3.2": 0.50, # Cheapest option
"gemini-2.5-flash": 0.30,
"gpt-4.1": 0.15,
"claude-sonnet-4.5": 0.05
}
def register_canary(self, name: str, target_config: Dict, initial_percentage: float = 5.0):
"""
Register a new canary deployment.
Args:
name: Unique identifier for this canary
target_config: {"model": "...", "temperature": ..., etc.}
initial_percentage: Starting traffic percentage (0-100)
"""
self.canary_configs[name] = {
"config": target_config,
"percentage": initial_percentage,
"total_requests": 0,
"successful_requests": 0,
"health_metrics": [],
"phase": "staging" # staging -> canary -> rollout -> complete
}
print(f"🎯 Canary '{name}' registered at {initial_percentage}% traffic")
def route_request(self, messages: List[Dict]) -> Dict[str, Any]:
"""
Route request based on current traffic split configuration.
Implements progressive canary rollout.
"""
# Check if any canary should handle this request
for name, config in self.canary_configs.items():
if config["phase"] in ["canary", "rollout"]:
if random.random() * 100 < config["percentage"]:
return self._execute_canary(name, messages)
# Default: use standard traffic split
return self._execute_default(messages)
def _execute_canary(self, name: str, messages: List[Dict]) -> Dict[str, Any]:
"""Execute request through a specific canary."""
config = self.canary_configs[name]
config["total_requests"] += 1
# Import here to avoid circular dependency
from ai_client import HolySheepAIClient
client = HolySheepAIClient()
result = client.chat_completion(
messages,
**config["config"]
)
if result["success"]:
config["successful_requests"] += 1
config["health_metrics"].append({
"timestamp": "now",
"success": True
})
self._evaluate_canary_health(name)
else:
config["health_metrics"].append({
"timestamp": "now",
"success": False,
"error": result.get("error")
})
result["canary"] = name
return result
def _execute_default(self, messages: List[Dict]) -> Dict[str, Any]:
"""Execute request using default traffic split."""
# Select model based on traffic weights
models = list(self.default_traffic_split.keys())
weights = list(self.default_traffic_split.values())
selected_model = random.choices(models, weights=weights)[0]
from ai_client import HolySheepAIClient
client = HolySheepAIClient()
return client.chat_completion(messages, model=selected_model)
def _evaluate_canary_health(self, name: str):
"""
Automatically scale canary based on health metrics.
If 95%+ success rate, increase traffic. If below 80%, decrease or pause.
"""
config = self.canary_configs[name]
if config["total_requests"] < 10:
return # Not enough data yet
success_rate = config["successful_requests"] / config["total_requests"]
if success_rate >= 0.95 and config["percentage"] < 50:
new_percentage = min(50, config["percentage"] + 10)
config["percentage"] = new_percentage
print(f"📈 Canary '{name}' success rate: {success_rate:.1%}, scaling to {new_percentage}%")
elif success_rate >= 0.98 and config["percentage"] >= 50:
config["phase"] = "complete"
print(f"🎉 Canary '{name}' fully rolled out at {config['percentage']}%")
elif success_rate < 0.80:
config["percentage"] = max(0, config["percentage"] - 5)
print(f"📉 Canary '{name}' health degraded: {success_rate:.1%}, scaling down to {config['percentage']}%")
if config["percentage"] <= 0:
config["phase"] = "paused"
print(f"⏸️ Canary '{name}' paused due to poor health")
Demonstration of canary release
if __name__ == "__main__":
canary = CanaryRelease(key_manager)
# Test new DeepSeek configuration
canary.register_canary(
name="deepseek-cost-opt",
target_config={"model": "deepseek-v3.2", "temperature": 0.5},
initial_percentage=5.0
)
# Simulate traffic
test_messages = [
{"role": "user", "content": "What is machine learning?"}
]
print("\n📊 Simulating 20 requests with canary active...")
for i in range(20):
result = canary.route_request(test_messages)
status = "✅" if result["success"] else "❌"
canary_info = f" (via {result.get('canary', 'default')})" if "canary" in result else ""
print(f"{status} Request {i + 1}{canary_info}")
print("\n📈 Canary Status:")
for name, config in canary.canary_configs.items():
success_rate = config["successful_requests"] / max(1, config["total_requests"])
print(f" {name}: {config['percentage']}% traffic, {success_rate:.1%} success rate")Step 5: Running the Complete System
Create a main entry point called main.py that ties everything together with scheduling for automatic key rotation.
import time
import schedule
from datetime import datetime
from key_manager import key_manager
from ai_client import HolySheepAIClient
from canary_release import CanaryRelease
def daily_health_check():
"""Run daily at midnight to check key health and rotate if needed."""
print("\n" + "=" * 60)
print(f"🔍 Daily Health Check - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
print(key_manager.get_status_report())
# Reset error counts for healthy keys
for key in key_manager.keys:
if key_manager.key_stats[key]['health_score'] > 80:
key_manager.key_stats[key]['errors'] = 0
print(f"✓ Reset error count for healthy key {key[:15]}...")
def hourly_rotation_check():
"""Hourly check to enforce rotation intervals."""
print(f"\n⏰ Hourly check at {datetime.now().strftime('%H:%M:%S')}")
time_since_rotation = (datetime.now() - key_manager.last_rotation).total_seconds() / 60
print(f" Minutes since last rotation: {time_since_rotation:.1f}")
def process_user_requests():
"""Example: Process real user requests with automatic rotation."""
client = HolySheepAIClient()
sample_requests = [
"Explain neural networks in simple terms",
"Write a Python function to sort a list",
"What are the benefits of API key rotation?"
]
for request in sample_requests:
print(f"\n💬 User: {request}")
messages = [
{"role": "system", "content": "You are a helpful technical assistant."},
{"role": "user", "content": request}
]
result = client.chat_completion(messages, model="deepseek-v3.2")
if result["success"]:
response = result["data"]["choices"][0]["message"]["content"]
print(f"🤖 HolySheep AI: {response[:100]}...")
else:
print(f"❌ Failed: {result.get('error')}")
def run_canary_demo():
"""Demonstrate canary release with cost optimization."""
canary = CanaryRelease(key_manager)
# Canary: Test Gemini Flash for simple queries (cheaper)
canary.register_canary(
name="gemini-fast-track",
target_config={
"model": "gemini-2.5-flash",
"temperature": 0.3,
"max_tokens": 500
},
initial_percentage=10.0
)
print("\n🧪 Canary Demo: Testing Gemini 2.5 Flash for simple queries")
print("-" * 50)
messages = [{"role": "user", "content": "What is 2+2?"}]
# Simulate 30 requests
gemini_hits = 0
for i in range(30):
result = canary.route_request(messages)
if result.get("canary") == "gemini-fast-track":
gemini_hits += 1
print(f"\n📊 Results: {gemini_hits}/30 requests went to Gemini 2.5 Flash")
print(f" Estimated savings: ${(gemini_hits * 0.001):.3f} vs GPT-4.1")
def main():
"""Main application entry point."""
print("\n" + "🏠" * 30)
print("HolySheep AI API Key Rotation System")
print("🏠" * 30)
print(f"\n✅ System initialized at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"✅ Loaded {len(key_manager.keys)} API keys")
# Schedule automatic tasks
schedule.every().day.at("00:00").do(daily_health_check)
schedule.every().hour.do(hourly_rotation_check)
# Run demonstration
print("\n" + "🎬" * 30)
print("DEMONSTRATION MODE")
print("🎬" * 30)
run_canary_demo()
print("\n" + "-" * 50)
process_user_requests()
# Final status
print("\n" + "📊" * 30)
print("Final System Status:")
print(key_manager.get_status_report())
print("\n💡 Next Steps:")
print(" 1. Add your HolySheep API keys to .env file")
print(" 2. Run 'python main.py' to test the system")
print(" 3. Customize traffic splits based on your needs")
print(" 4. Set up Redis for distributed key management")
if __name__ == "__main__":
main()[Screenshot hint: Terminal shows colored output with emojis indicating system status and request routing]
How to Calculate Your Cost Savings
Let's say your application makes 1 million API calls monthly, averaging 500 tokens output each:
- Without rotation (all GPT-4.1): 1,000,000 × 500 / 1,000,000 × $8 = $4,000/month
- With intelligent routing (50% DeepSeek V3.2, 30% Gemini Flash, 20% GPT-4.1):
- DeepSeek: 500,000 × 500 / 1,000,000 × $0.42 = $105
- Gemini: 300,000 × 500 / 1,000,000 × $2.50 = $375
- GPT-4.1: 200,000 × 500 / 1,000,000 × $8 = $800
- Your monthly savings: $2,720 (68% reduction)
With HolySheep's ¥1 = $1 rate instead of ¥7.3 per dollar, your effective savings jump to 85%+ compared to standard pricing.
Common Errors and Fixes
Error 1: "No API keys found" on Startup
Problem: Your application crashes immediately with ValueError: No API keys found.
Cause: The .env file is missing, incorrectly named, or API keys aren't prefixed correctly.
# ❌ WRONG - Missing or wrong prefix
HOLYSHEEP_API_KEY=sk_live_abc123
❌ WRONG - Keys not being loaded
my_key=hs_live_abc123
✅ CORRECT - Proper prefix and underscore
HOLYSHEEP_KEY_1=hs_live_abc123def456
HOLYSHEEP_KEY_2=hs_live_xyz789ghi012Fix: Create a properly formatted .env file in your project root:
# Create .env file with proper naming
touch .env # On Mac/Linux
Or create manually on Windows
Add these exact lines:
HOLYSHEEP_KEY_1=hs_live_your_first_key_here
HOLYSHEEP_KEY_2=hs_live_your_second_key_here
HOLYSHEEP_KEY_3=hs_live_your_third_key_here
Verify the file is in your project root
ls -la .env # Should show the file
Test loading
python -c "from dotenv import load_dotenv; load_dotenv(); import os; print(os.getenv('HOLYSHEEP_KEY_1'))"Error 2: HTTP 401 Authentication Error
Problem: All requests fail with HTTP 401 Unauthorized even with valid-looking keys.
Cause: Keys are expired, revoked, or the Authorization header format is incorrect.
# ❌ WRONG - Wrong authorization format
headers = {
"Authorization": api_key, # Missing "Bearer " prefix
}
✅ CORRECT - Proper Bearer token format
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}Fix: Verify your keys in the HolySheep dashboard and ensure correct header format:
# Test API key directly with curl
curl -X POST https://api.holysheep.ai/v1/chat/completions \
-H "Authorization: Bearer YOUR_HOLYSHEEP_KEY_1" \
-H "Content-Type: application/json" \
-d '{"model":"deepseek-v3.2","messages":[{"role":"user","content":"test"}]}'
If you get {"error": {"code": "invalid_api_key"}}, the key is invalid
If you get {"error": {"code": "insufficient_quota"}}, your credits are exhausted
Check your dashboard at https://www.holysheep.ai/dashboard
Error 3: Infinite Retry Loop During Rate Limiting
Problem: Your application hangs indefinitely when hitting rate limits, never failing gracefully.
Cause: The retry logic doesn't have a maximum attempts limit or doesn't distinguish between retryable and non-retryable errors.
# ❌ WRONG - No max retries, potential infinite loop
def chat_completion(self, messages):
while True:
response = requests.post(url, headers=headers)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
time.sleep(1) # Could loop foreverFix: Implement proper retry logic with exponential backoff and clear error codes:
import requests
from requests.exceptions import Timeout, ConnectionError
class FixedRetryClient:
def __init__(self, max_retries=3):
self.max_retries = max_retries
def chat_completion(self, messages):
for attempt in range(self.max_retries):
try:
response = requests.post(url, headers=headers, timeout=30)
# Success
if response.status_code == 200:
return response.json()
# Rate limited - retry with backoff
if response.status_code == 429:
wait_time = 2 ** attempt # 1, 2, 4 seconds
print(f"Rate limited. Waiting {wait_time}s before retry...")
time.sleep(wait_time)
continue
# Authentication error - don't retry, will never succeed
if response.status_code == 401:
return {"error": "Invalid API key", "code": "AUTH_FAILED"}
# Server error - might succeed on retry
if response.status_code >= 500:
time.sleep(2 ** attempt)
continue
# Client error - won't succeed with retry
return {"error": f"Client error: {response.status_code}"}
except Timeout:
print(f"Request timeout. Retry {attempt + 1}/{self.max_retries}")
time.sleep(2 ** attempt)
except ConnectionError as e:
print(f"Connection failed: {e}")
time.sleep(5) # Longer wait for connection issues
return {"error": "Max retries exceeded", "code": "RETRIES_EXHAUSTED"}Error 4: Key Health Score Never Recovering
Problem: Once a key's health score drops, it never recovers, even after successful requests.
Cause: The health recovery rate is too slow or disabled, or error penalties are too severe.
# ❌ WRONG - Health never recovers
def report_success(self, key):
pass # Does nothing to recover health
def report_error(self, key):
self.key_stats[key]['health_score'] -= 20 # Too harshFix: Implement balanced recovery logic:
def report_success(self, key):
"""Gradually recover health after successful requests."""
current = self.key_stats[key]['health_score']
recovery = 2.0 if current > 50 else 4.0 # Faster recovery for degraded keys
self.key_stats[key]['health_score'] = min(100, current + recovery)
def report_error(self, key, error_type=None):
"""Apply graduated penalties based on error severity."""
penalties = {
'rate_limit': 15, # External service limit
'timeout': 10, # Network issue
'server_error': 8, # Might recover
'auth_error': 100, # Key is invalid - remove immediately
'unknown_error': 5 # Minor penalty
}
penalty = penalties.get(error_type, 5)
self.key_stats[key]['health_score'] = max(0, self.key_stats[key]['health_score'] - penalty)
# Remove key if health is critically low
if self.key_stats[key]['health_score'] <= 0:
self._mark_key_failed(key)
Add recovery baseline in daily health check
def daily_health_recovery(self):
"""Prevent keys from getting permanently stuck at low health."""
for key in self.keys:
stats = self.key_stats[key]
# Allow partial recovery each day even without requests
if stats['errors'] == 0 and stats['health_score'] < 100:
stats['