Coverage for apps / ai / services / openrouter.py: 20%
141 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 00:40 +0000
1"""OpenRouter API service using the official SDK."""
3import json
4import logging
5import time
6from typing import Any
8from openrouter import OpenRouter
10from apps.core.models import AppSettings
12logger = logging.getLogger(__name__)
15class AIServiceError(Exception):
16 """Base exception for AI service errors."""
17 pass
20class AIUnavailableError(AIServiceError):
21 """Raised when AI service is not available (no API key)."""
22 pass
25class AIResponseError(AIServiceError):
26 """Raised when AI returns an invalid or unexpected response."""
27 pass
30class OpenRouterService:
31 """Service for interacting with OpenRouter API."""
33 # Class-level cache for API key validation: {key_hash: (is_valid, timestamp)}
34 _key_validation_cache: dict[int, tuple[bool, float]] = {}
35 KEY_VALIDATION_TTL = 300 # 5 minutes
37 def __init__(self, api_key: str | None = None):
38 """Initialize the service with an API key.
40 Args:
41 api_key: OpenRouter API key. If None, fetches from AppSettings.
42 """
43 if api_key is None:
44 settings = AppSettings.get()
45 api_key = settings.openrouter_api_key
47 if not api_key:
48 raise AIUnavailableError('OpenRouter API key not configured')
50 self.api_key = api_key
52 def complete(
53 self,
54 system_prompt: str,
55 user_prompt: str,
56 model: str = 'anthropic/claude-3.5-haiku',
57 json_response: bool = True,
58 ) -> dict[str, Any]:
59 """Send a completion request to OpenRouter.
61 Args:
62 system_prompt: System message for the AI.
63 user_prompt: User message/query.
64 model: Model identifier (e.g., 'anthropic/claude-3.5-haiku').
65 json_response: Whether to request JSON output.
67 Returns:
68 Parsed JSON response from the AI, or raw text response.
70 Raises:
71 AIUnavailableError: If no API key is configured.
72 AIResponseError: If the response is invalid.
73 """
74 messages = [
75 {'role': 'system', 'content': system_prompt},
76 {'role': 'user', 'content': user_prompt},
77 ]
79 try:
80 with OpenRouter(api_key=self.api_key) as client:
81 response = client.chat.send(
82 messages=messages,
83 model=model,
84 stream=False,
85 )
87 # Extract the response content
88 if not response or not hasattr(response, 'choices'):
89 raise AIResponseError('Invalid response structure from OpenRouter')
91 if not response.choices:
92 raise AIResponseError('No choices in OpenRouter response')
94 content = response.choices[0].message.content
96 if json_response:
97 # Parse JSON from the response
98 try:
99 # Handle potential markdown code blocks
100 if content.startswith('```'):
101 # Extract JSON from code block
102 lines = content.split('\n')
103 json_lines = []
104 in_block = False
105 for line in lines:
106 if line.startswith('```'):
107 in_block = not in_block
108 continue
109 if in_block:
110 json_lines.append(line)
111 content = '\n'.join(json_lines)
113 return json.loads(content)
114 except json.JSONDecodeError as e:
115 # Note: This log appears in test output for test_complete_invalid_json - expected
116 logger.error(f'Failed to parse AI response as JSON: {content}')
117 raise AIResponseError(f'Invalid JSON in AI response: {e}')
119 return {'content': content}
121 except AIServiceError:
122 raise
123 except Exception as e:
124 logger.exception('OpenRouter API error')
125 raise AIResponseError(f'OpenRouter API error: {e}')
127 async def complete_async(
128 self,
129 system_prompt: str,
130 user_prompt: str,
131 model: str = 'anthropic/claude-3.5-haiku',
132 json_response: bool = True,
133 ) -> dict[str, Any]:
134 """Async version of complete().
136 Args:
137 system_prompt: System message for the AI.
138 user_prompt: User message/query.
139 model: Model identifier (e.g., 'anthropic/claude-3.5-haiku').
140 json_response: Whether to request JSON output.
142 Returns:
143 Parsed JSON response from the AI, or raw text response.
144 """
145 messages = [
146 {'role': 'system', 'content': system_prompt},
147 {'role': 'user', 'content': user_prompt},
148 ]
150 try:
151 async with OpenRouter(api_key=self.api_key) as client:
152 response = await client.chat.send_async(
153 messages=messages,
154 model=model,
155 stream=False,
156 )
158 # Extract the response content
159 if not response or not hasattr(response, 'choices'):
160 raise AIResponseError('Invalid response structure from OpenRouter')
162 if not response.choices:
163 raise AIResponseError('No choices in OpenRouter response')
165 content = response.choices[0].message.content
167 if json_response:
168 # Parse JSON from the response
169 try:
170 # Handle potential markdown code blocks
171 if content.startswith('```'):
172 lines = content.split('\n')
173 json_lines = []
174 in_block = False
175 for line in lines:
176 if line.startswith('```'):
177 in_block = not in_block
178 continue
179 if in_block:
180 json_lines.append(line)
181 content = '\n'.join(json_lines)
183 return json.loads(content)
184 except json.JSONDecodeError as e:
185 logger.error(f'Failed to parse AI response as JSON: {content}')
186 raise AIResponseError(f'Invalid JSON in AI response: {e}')
188 return {'content': content}
190 except AIServiceError:
191 raise
192 except Exception as e:
193 logger.exception('OpenRouter API error')
194 raise AIResponseError(f'OpenRouter API error: {e}')
196 @classmethod
197 def is_available(cls) -> bool:
198 """Check if AI service is available (API key configured)."""
199 settings = AppSettings.get()
200 return bool(settings.openrouter_api_key)
202 def get_available_models(self) -> list[dict[str, str]]:
203 """Get list of available models from OpenRouter.
205 Returns:
206 List of dicts with 'id' and 'name' keys for each available model.
208 Raises:
209 AIResponseError: If the API call fails.
210 """
211 try:
212 with OpenRouter(api_key=self.api_key) as client:
213 response = client.models.list()
215 if not response or not hasattr(response, 'data'):
216 raise AIResponseError('Invalid response from OpenRouter models API')
218 models = [
219 {'id': model.id, 'name': model.name}
220 for model in response.data
221 if hasattr(model, 'id') and hasattr(model, 'name')
222 ]
223 return sorted(models, key=lambda m: m['name'].lower())
224 except AIServiceError:
225 raise
226 except Exception as e:
227 logger.exception('Failed to fetch OpenRouter models')
228 raise AIResponseError(f'Failed to fetch available models: {e}')
230 @classmethod
231 def test_connection(cls, api_key: str) -> tuple[bool, str]:
232 """Test if an API key is valid by making a minimal request.
234 Args:
235 api_key: The API key to test.
237 Returns:
238 Tuple of (success: bool, message: str)
239 """
240 try:
241 service = cls(api_key=api_key)
242 # Make a minimal test request
243 service.complete(
244 system_prompt='Respond with exactly: {"status": "ok"}',
245 user_prompt='Test connection',
246 model='anthropic/claude-3.5-haiku',
247 json_response=True,
248 )
249 return True, 'Connection successful'
250 except AIUnavailableError:
251 return False, 'API key not provided'
252 except AIResponseError as e:
253 return False, f'API error: {e}'
254 except Exception as e:
255 return False, f'Connection failed: {e}'
257 @classmethod
258 def validate_key_cached(cls, api_key: str | None = None) -> tuple[bool, str | None]:
259 """Validate API key with caching to avoid excessive API calls.
261 Args:
262 api_key: API key to validate. If None, fetches from AppSettings.
264 Returns:
265 Tuple of (is_valid: bool, error_message: str | None)
266 """
267 if api_key is None:
268 settings = AppSettings.get()
269 api_key = settings.openrouter_api_key
271 if not api_key:
272 return False, 'No API key configured'
274 key_hash = hash(api_key)
275 now = time.time()
277 # Check cache
278 if key_hash in cls._key_validation_cache:
279 is_valid, timestamp = cls._key_validation_cache[key_hash]
280 if now - timestamp < cls.KEY_VALIDATION_TTL:
281 return is_valid, None if is_valid else 'API key is invalid or expired'
283 # Validate with API
284 try:
285 is_valid, message = cls.test_connection(api_key)
286 cls._key_validation_cache[key_hash] = (is_valid, now)
287 return is_valid, None if is_valid else message
288 except Exception as e:
289 logger.exception('Failed to validate API key')
290 return False, f'Unable to verify API key: {e}'
292 @classmethod
293 def invalidate_key_cache(cls):
294 """Clear validation cache (call when key is updated)."""
295 cls._key_validation_cache.clear()