Coverage for src/openapi_navigator/spec_manager.py: 76%
218 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-15 01:16 +0000
1"""OpenAPI Navigator - Specification manager for loading, validating, and indexing specs."""
3import json
4import logging
5from pathlib import Path
6from typing import Dict, List, Optional, Any
7import yaml
8import requests
9from jsonschema import validate, ValidationError
11logger = logging.getLogger(__name__)
14class OpenAPISpec:
15 """Represents a loaded OpenAPI specification with indexes for fast lookups."""
17 def __init__(self, spec_id: str, spec_data: Dict[str, Any]):
18 self.spec_id = spec_id
19 self.spec_data = spec_data
20 self.version = self._detect_version()
21 self.default_headers: Dict[str, str] = {}
22 self._build_indexes()
24 def _detect_version(self) -> str:
25 """Detect if this is OpenAPI 3.x or Swagger 2.x."""
26 if "openapi" in self.spec_data:
27 return "openapi3"
28 elif "swagger" in self.spec_data:
29 return "swagger2"
30 else:
31 raise ValueError("Could not determine OpenAPI/Swagger version")
33 def _build_indexes(self) -> None:
34 """Build fast lookup indexes for endpoints and schemas."""
35 self.endpoints = []
36 self.schemas = {}
38 # Extract endpoints
39 if self.version == "openapi3":
40 paths = self.spec_data.get("paths", {})
41 logger.info(
42 f"Building indexes for OpenAPI 3.x spec with {len(paths)} paths"
43 )
44 for path, path_item in paths.items():
45 for method, operation in path_item.items():
46 if method in [
47 "get",
48 "post",
49 "put",
50 "delete",
51 "patch",
52 "head",
53 "options",
54 "trace",
55 ]:
56 self.endpoints.append(
57 {
58 "method": method.upper(),
59 "path": path,
60 "summary": operation.get("summary", ""),
61 "operationId": operation.get("operationId", ""),
62 "tags": operation.get("tags", []),
63 "operation": operation,
64 }
65 )
66 else: # swagger2
67 paths = self.spec_data.get("paths", {})
68 logger.info(
69 f"Building indexes for Swagger 2.x spec with {len(paths)} paths"
70 )
71 for path, path_item in paths.items():
72 for method, operation in path_item.items():
73 if method in [
74 "get",
75 "post",
76 "put",
77 "delete",
78 "patch",
79 "head",
80 "options",
81 ]:
82 self.endpoints.append(
83 {
84 "method": method.upper(),
85 "path": path,
86 "summary": operation.get("summary", ""),
87 "operationId": operation.get("operationId", ""),
88 "tags": operation.get("tags", []),
89 "operation": operation,
90 }
91 )
93 # Extract schemas
94 if self.version == "openapi3":
95 schemas = self.spec_data.get("components", {}).get("schemas", {})
96 else: # swagger2
97 schemas = self.spec_data.get("definitions", {})
99 self.schemas = schemas
101 logger.info(
102 f"Built indexes: {len(self.endpoints)} endpoints, {len(self.schemas)} schemas"
103 )
104 if self.endpoints:
105 logger.info(f"Sample endpoint: {self.endpoints[0]}")
106 if self.schemas:
107 logger.info(f"Sample schema names: {list(self.schemas.keys())[:5]}")
109 def get_endpoint(self, path: str, method: str) -> Optional[Dict[str, Any]]:
110 """Get a specific endpoint by path and method."""
111 for endpoint in self.endpoints:
112 if endpoint["path"] == path and endpoint["method"] == method:
113 operation = endpoint["operation"]
115 # Special handling for root endpoint - provide condensed view to reduce token usage
116 if path == "/" and operation:
117 # Check if this is a very large operation (likely contains full API documentation)
118 operation_str = str(operation)
119 if len(operation_str) > 5000: # If operation is very large
120 return {
121 "summary": operation.get("summary", "API Root Endpoint"),
122 "description": operation.get("description", "")[:500]
123 + (
124 "..."
125 if len(operation.get("description", "")) > 500
126 else ""
127 ),
128 "operationId": operation.get("operationId", ""),
129 "tags": operation.get("tags", []),
130 "_note": "This is a condensed view of the root endpoint. Use summary_only=False on get_endpoint for full details.",
131 "responses": {
132 code: {
133 "description": resp.get("description", "")[:200]
134 + (
135 "..."
136 if len(resp.get("description", "")) > 200
137 else ""
138 ),
139 "content_types": list(
140 resp.get("content", {}).keys()
141 )
142 if "content" in resp
143 else [],
144 }
145 for code, resp in list(
146 operation.get("responses", {}).items()
147 )[:3] # Limit to first 3 responses
148 },
149 }
151 return operation
152 return None
154 def list_endpoints(self, tag: Optional[str] = None) -> List[Dict[str, Any]]:
155 """List endpoints, optionally filtered by tag."""
156 logger.info(
157 f"list_endpoints called with tag={tag}, self.endpoints has {len(self.endpoints)} items"
158 )
159 if tag is None:
160 result = [
161 {
162 "method": e["method"],
163 "path": e["path"],
164 "summary": e["summary"],
165 "operationId": e["operationId"],
166 }
167 for e in self.endpoints
168 ]
169 logger.info(f"Returning {len(result)} endpoints (no tag filter)")
170 return result
172 filtered = []
173 for endpoint in self.endpoints:
174 if tag in endpoint["tags"]:
175 filtered.append(
176 {
177 "method": endpoint["method"],
178 "path": endpoint["path"],
179 "summary": endpoint["summary"],
180 "operationId": endpoint["operationId"],
181 }
182 )
183 logger.info(f"Returning {len(filtered)} endpoints filtered by tag '{tag}'")
184 return filtered
186 def search_endpoints(
187 self, query: str, limit: int = 50, offset: int = 0
188 ) -> Dict[str, Any]:
189 """Search endpoints using fuzzy matching with pagination.
191 Args:
192 query: Search query string. Use "" or "a" to get all endpoints.
193 limit: Maximum number of results to return (default 50, max 200)
194 offset: Number of results to skip for pagination (default 0)
196 Returns:
197 Dictionary containing:
198 - endpoints: List of matching endpoints
199 - total: Total number of matches (before pagination)
200 - limit: Applied limit
201 - offset: Applied offset
202 - has_more: Whether there are more results available
203 """
204 # Validate and clamp limit
205 limit = max(1, min(limit, 200))
206 offset = max(0, offset)
208 # If query is empty or very short, return all endpoints with high scores
209 if not query or len(query.strip()) <= 2:
210 all_results = []
211 for endpoint in self.endpoints:
212 all_results.append(
213 {
214 "method": endpoint["method"],
215 "path": endpoint["path"],
216 "summary": endpoint["summary"],
217 "operationId": endpoint["operationId"],
218 "score": 100,
219 }
220 )
222 # Apply pagination
223 total = len(all_results)
224 paginated_results = all_results[offset : offset + limit]
226 return {
227 "endpoints": paginated_results,
228 "total": total,
229 "limit": limit,
230 "offset": offset,
231 "has_more": offset + limit < total,
232 }
234 try:
235 from fuzzywuzzy import fuzz
236 except ImportError:
237 # Fallback to simple substring search if fuzzywuzzy is not available
238 all_results = []
239 query_lower = query.lower()
240 for endpoint in self.endpoints:
241 if (
242 query_lower in endpoint["path"].lower()
243 or query_lower in endpoint["summary"].lower()
244 or query_lower in endpoint["operationId"].lower()
245 ):
246 all_results.append(
247 {
248 "method": endpoint["method"],
249 "path": endpoint["path"],
250 "summary": endpoint["summary"],
251 "operationId": endpoint["operationId"],
252 "score": 100,
253 }
254 )
256 # Apply pagination
257 total = len(all_results)
258 paginated_results = all_results[offset : offset + limit]
260 return {
261 "endpoints": paginated_results,
262 "total": total,
263 "limit": limit,
264 "offset": offset,
265 "has_more": offset + limit < total,
266 }
268 all_results = []
269 for endpoint in self.endpoints:
270 # Search in path, summary, operationId
271 path_score = fuzz.partial_ratio(query.lower(), endpoint["path"].lower())
272 summary_score = fuzz.partial_ratio(
273 query.lower(), endpoint["summary"].lower()
274 )
275 op_id_score = fuzz.partial_ratio(
276 query.lower(), endpoint["operationId"].lower()
277 )
279 # Use the best score
280 best_score = max(path_score, summary_score, op_id_score)
282 if best_score > 50: # Threshold for relevance
283 all_results.append(
284 {
285 "method": endpoint["method"],
286 "path": endpoint["path"],
287 "summary": endpoint["summary"],
288 "operationId": endpoint["operationId"],
289 "score": best_score,
290 }
291 )
293 # Sort by relevance score
294 all_results.sort(key=lambda x: x["score"], reverse=True)
296 # Apply pagination
297 total = len(all_results)
298 paginated_results = all_results[offset : offset + limit]
300 return {
301 "endpoints": paginated_results,
302 "total": total,
303 "limit": limit,
304 "offset": offset,
305 "has_more": offset + limit < total,
306 }
308 def search_schemas(
309 self, query: str, limit: int = 50, offset: int = 0
310 ) -> Dict[str, Any]:
311 """Search schema names using fuzzy matching with pagination.
313 Args:
314 query: Search query string. Use "" or "a" to get all schemas.
315 limit: Maximum number of results to return (default 50, max 200)
316 offset: Number of results to skip for pagination (default 0)
318 Returns:
319 Dictionary containing:
320 - schemas: List of matching schema names
321 - total: Total number of matches (before pagination)
322 - limit: Applied limit
323 - offset: Applied offset
324 - has_more: Whether there are more results available
325 """
326 # Validate and clamp limit
327 limit = max(1, min(limit, 200))
328 offset = max(0, offset)
330 # If query is empty or very short, return all schemas with high scores
331 if not query or len(query.strip()) <= 2:
332 all_results = []
333 for schema_name in self.schemas.keys():
334 all_results.append({"name": schema_name, "score": 100})
336 # Apply pagination
337 total = len(all_results)
338 paginated_results = all_results[offset : offset + limit]
340 return {
341 "schemas": paginated_results,
342 "total": total,
343 "limit": limit,
344 "offset": offset,
345 "has_more": offset + limit < total,
346 }
348 try:
349 from fuzzywuzzy import fuzz
350 except ImportError:
351 # Fallback to simple substring search if fuzzywuzzy is not available
352 all_results = []
353 query_lower = query.lower()
354 for schema_name in self.schemas.keys():
355 if query_lower in schema_name.lower():
356 all_results.append({"name": schema_name, "score": 100})
358 # Apply pagination
359 total = len(all_results)
360 paginated_results = all_results[offset : offset + limit]
362 return {
363 "schemas": paginated_results,
364 "total": total,
365 "limit": limit,
366 "offset": offset,
367 "has_more": offset + limit < total,
368 }
370 all_results = []
371 for schema_name in self.schemas.keys():
372 # Search in schema name
373 score = fuzz.partial_ratio(query.lower(), schema_name.lower())
375 if score > 50: # Threshold for relevance
376 all_results.append({"name": schema_name, "score": score})
378 # Sort by relevance score
379 all_results.sort(key=lambda x: x["score"], reverse=True)
381 # Apply pagination
382 total = len(all_results)
383 paginated_results = all_results[offset : offset + limit]
385 return {
386 "schemas": paginated_results,
387 "total": total,
388 "limit": limit,
389 "offset": offset,
390 "has_more": offset + limit < total,
391 }
393 def get_schema(self, schema_name: str) -> Optional[Dict[str, Any]]:
394 """Get a specific schema by name."""
395 return self.schemas.get(schema_name)
397 def get_spec_metadata(self) -> Dict[str, Any]:
398 """Get metadata about the OpenAPI specification."""
399 metadata = {
400 "spec_id": self.spec_id,
401 "version": self.version,
402 "openapi_version": self.spec_data.get(
403 "openapi", self.spec_data.get("swagger", "unknown")
404 ),
405 "title": self.spec_data.get("info", {}).get("title", ""),
406 "description": self.spec_data.get("info", {}).get("description", ""),
407 "version_info": self.spec_data.get("info", {}).get("version", ""),
408 "contact": self.spec_data.get("info", {}).get("contact", {}),
409 "license": self.spec_data.get("info", {}).get("license", {}),
410 "terms_of_service": self.spec_data.get("info", {}).get(
411 "termsOfService", ""
412 ),
413 "servers": self.spec_data.get("servers", []),
414 "host": self.spec_data.get("host", ""),
415 "base_path": self.spec_data.get("basePath", ""),
416 "schemes": self.spec_data.get("schemes", []),
417 "tags": self.spec_data.get("tags", []),
418 "endpoint_count": len(self.endpoints),
419 "schema_count": len(self.schemas),
420 }
421 return metadata
423 def list_schemas(self, prefix: Optional[str] = None) -> List[str]:
424 """List schema names, optionally filtered by prefix."""
425 logger.info(
426 f"list_schemas called with prefix={prefix}, self.schemas has {len(self.schemas)} items"
427 )
428 schema_names = list(self.schemas.keys())
429 if prefix:
430 schema_names = [
431 name for name in schema_names if name.lower().startswith(prefix.lower())
432 ]
433 logger.info(
434 f"Returning {len(schema_names)} schema names (filtered by prefix '{prefix}')"
435 )
436 return sorted(schema_names)
438 def set_headers(self, headers: Dict[str, str]) -> None:
439 """Set default headers for API requests using this spec.
441 Args:
442 headers: Dictionary of HTTP headers to use by default
443 """
444 self.default_headers = headers.copy() if headers else {}
446 def get_headers(self) -> Dict[str, str]:
447 """Get default headers for API requests using this spec.
449 Returns:
450 Copy of the default headers dictionary
451 """
452 return self.default_headers.copy()
455class SpecManager:
456 """Manages multiple loaded OpenAPI specifications."""
458 def __init__(self):
459 self.specs: Dict[str, OpenAPISpec] = {}
460 self._openapi3_schema = None
461 self._swagger2_schema = None
463 def _get_openapi3_schema(self) -> Dict[str, Any]:
464 """Get OpenAPI 3.x JSON schema for validation."""
465 if self._openapi3_schema is None:
466 # Basic OpenAPI 3.x schema - we could load the full one from json-schema.org
467 self._openapi3_schema = {
468 "type": "object",
469 "required": ["openapi", "info", "paths"],
470 "properties": {
471 "openapi": {"type": "string"},
472 "info": {"type": "object"},
473 "paths": {"type": "object"},
474 },
475 }
476 return self._openapi3_schema
478 def _get_swagger2_schema(self) -> Dict[str, Any]:
479 """Get Swagger 2.x JSON schema for validation."""
480 if self._swagger2_schema is None:
481 # Basic Swagger 2.x schema
482 self._swagger2_schema = {
483 "type": "object",
484 "required": ["swagger", "info", "paths"],
485 "properties": {
486 "swagger": {"type": "string"},
487 "info": {"type": "object"},
488 "paths": {"type": "object"},
489 },
490 }
491 return self._swagger2_schema
493 def load_spec_from_file(self, file_path: str, spec_id: Optional[str] = None) -> str:
494 """Load an OpenAPI spec from a local file."""
495 if not Path(file_path).is_absolute():
496 raise ValueError("File path must be absolute")
498 if not Path(file_path).exists():
499 raise FileNotFoundError(f"File not found: {file_path}")
501 try:
502 with open(file_path, "r", encoding="utf-8") as f:
503 if file_path.endswith((".yaml", ".yml")):
504 spec_data = yaml.safe_load(f)
505 else:
506 spec_data = json.load(f)
508 return self._load_spec(spec_data, spec_id or f"file:{file_path}")
510 except Exception as e:
511 logger.error(f"Failed to load spec from file {file_path}: {e}")
512 raise
514 def load_spec_from_url(self, url: str, spec_id: Optional[str] = None) -> str:
515 """Load an OpenAPI spec from a URL."""
516 try:
517 response = requests.get(url, timeout=30)
518 response.raise_for_status()
520 # Robust format detection: try JSON first, fall back to YAML
521 try:
522 # Try JSON first (most common and fastest to parse)
523 spec_data = response.json()
524 except (ValueError, TypeError):
525 # If JSON fails, try YAML
526 try:
527 spec_data = yaml.safe_load(response.text)
528 except yaml.YAMLError as e:
529 raise ValueError(f"Failed to parse as JSON or YAML: {e}")
531 return self._load_spec(spec_data, spec_id or f"url:{url}")
533 except Exception as e:
534 logger.error(f"Failed to load spec from URL {url}: {e}")
535 raise
537 def _load_spec(self, spec_data: Dict[str, Any], spec_id: str) -> str:
538 """Internal method to load and validate a spec."""
539 try:
540 # Basic validation
541 if "openapi" in spec_data:
542 validate(spec_data, self._get_openapi3_schema())
543 elif "swagger" in spec_data:
544 validate(spec_data, self._get_swagger2_schema())
545 else:
546 raise ValueError(
547 "Spec does not appear to be a valid OpenAPI or Swagger document"
548 )
550 # Create the spec object
551 spec = OpenAPISpec(spec_id, spec_data)
552 self.specs[spec_id] = spec
554 logger.info(
555 f"Successfully loaded spec {spec_id} with {len(spec.endpoints)} endpoints and {len(spec.schemas)} schemas"
556 )
557 return spec_id
559 except ValidationError as e:
560 logger.warning(f"Spec validation warnings for {spec_id}: {e}")
561 # Continue anyway - just warn
562 spec = OpenAPISpec(spec_id, spec_data)
563 self.specs[spec_id] = spec
564 return spec_id
565 except Exception as e:
566 logger.error(f"Failed to load spec {spec_id}: {e}")
567 raise
569 def get_spec(self, spec_id: str) -> Optional[OpenAPISpec]:
570 """Get a loaded spec by ID."""
571 return self.specs.get(spec_id)
573 def list_loaded_specs(self) -> List[str]:
574 """List IDs of all loaded specs."""
575 return list(self.specs.keys())
577 def unload_spec(self, spec_id: str) -> bool:
578 """Unload a spec by ID."""
579 if spec_id in self.specs:
580 del self.specs[spec_id]
581 logger.info(f"Unloaded spec {spec_id}")
582 return True
583 return False