File size: 5,966 Bytes
b30f068 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | """
API Router
Routes API calls to the appropriate client based on API type.
Handles initialization of clients with proper authentication.
"""
from typing import Dict, Any, Optional
import sys
from pathlib import Path
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from api_clients.weather_client import WeatherAPIClient
from config.credentials import get_api_key
class APIRouter:
"""
Route API calls to the appropriate client.
Manages client initialization and routes requests based on API type.
"""
def __init__(self):
"""Initialize the API router with available clients."""
self.clients = {}
self._initialize_clients()
def _initialize_clients(self):
"""Initialize all available API clients."""
# Initialize Weather API client
try:
weather_api_key = get_api_key("OPENWEATHER_API_KEY")
if weather_api_key:
self.clients['weather'] = WeatherAPIClient(api_key=weather_api_key)
print("✅ Weather API client initialized")
else:
print("⚠️ Weather API key not found - weather queries will not work")
except Exception as e:
print(f"❌ Failed to initialize Weather API client: {e}")
# Add more clients here as they are implemented
# Example:
# try:
# openai_api_key = get_api_key("OPENAI_API_KEY")
# if openai_api_key:
# self.clients['openai'] = OpenAIClient(api_key=openai_api_key)
# except Exception as e:
# print(f"Failed to initialize OpenAI client: {e}")
def call_api(
self,
api_type: str,
params: Dict[str, Any],
api_name: Optional[str] = None
) -> Dict[str, Any]:
"""
Route the API call to the correct client.
Args:
api_type: Type of API (e.g., 'weather', 'openai', 'usda')
params: Parameters for the API call
api_name: Optional specific API name for logging
Returns:
API response dictionary
Example:
>>> router = APIRouter()
>>> response = router.call_api('weather', {'city': 'London'})
>>> if response['success']:
... print(f"Temperature: {response['temperature']}°C")
"""
# Check if API type is supported
if api_type not in self.clients:
return {
"success": False,
"error": f"API type '{api_type}' not available",
"message": f"❌ {api_type.title()} API is not configured or not available",
"available_apis": list(self.clients.keys())
}
# Get the appropriate client
client = self.clients[api_type]
# Make the API call
try:
print(f"📡 Calling {api_type} API with params: {params}")
response = client.call(params)
if response.get('success'):
print(f"✅ {api_type} API call successful")
else:
print(f"⚠️ {api_type} API call failed: {response.get('message')}")
return response
except Exception as e:
error_msg = f"Failed to call {api_type} API: {str(e)}"
print(f"❌ {error_msg}")
return {
"success": False,
"error": str(e),
"message": f"❌ {error_msg}"
}
def is_api_available(self, api_type: str) -> bool:
"""
Check if an API type is available.
Args:
api_type: Type of API to check
Returns:
True if API is initialized and available
"""
return api_type in self.clients
def get_available_apis(self) -> list:
"""
Get list of available API types.
Returns:
List of API type strings
"""
return list(self.clients.keys())
def get_api_status(self) -> Dict[str, bool]:
"""
Get status of all API clients.
Returns:
Dictionary mapping API types to their availability status
"""
return {api_type: True for api_type in self.clients.keys()}
def close_all(self):
"""Close all API client sessions."""
for client in self.clients.values():
if hasattr(client, 'close'):
client.close()
# Singleton instance for global use
_router_instance = None
def get_router() -> APIRouter:
"""
Get the singleton APIRouter instance.
Returns:
APIRouter instance
"""
global _router_instance
if _router_instance is None:
_router_instance = APIRouter()
return _router_instance
# Test the router if run directly
if __name__ == "__main__":
print("🔀 Testing API Router\n")
# Create router
router = APIRouter()
# Check available APIs
print(f"Available APIs: {router.get_available_apis()}\n")
# Test weather API if available
if router.is_api_available('weather'):
print("🌤️ Testing Weather API")
print("-" * 50)
test_params = {
'city': 'London',
'units': 'metric'
}
result = router.call_api('weather', test_params)
if result.get('success'):
print(f"\n✅ Success!")
print(f"Location: {result.get('location')}")
print(f"Temperature: {result.get('temperature')}°C")
print(f"Description: {result.get('description')}")
else:
print(f"\n❌ Error: {result.get('message')}")
else:
print("⚠️ Weather API not available")
# Clean up
router.close_all()
|