Initial upload: mockapi - OpenAPI Mock Server Generator

src/mockapi/core/spec_loader.py

@@ -0,0 +1,93 @@
+"""OpenAPI Specification Loader."""
+
+import json
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import yaml
+
+
+class SpecLoaderError(Exception):
+    """Raised when spec loading fails."""
+    pass
+
+
+class SpecLoader:
+    """Loads and parses OpenAPI specifications from YAML or JSON files."""
+
+    SUPPORTED_EXTENSIONS = {".yaml", ".yml", ".json"}
+
+    def __init__(self, spec_path: str, fmt: Optional[str] = None):
+        """Initialize the spec loader.
+
+        Args:
+            spec_path: Path to the OpenAPI spec file
+            fmt: Force format (yaml or json), auto-detected if None
+        """
+        self.spec_path = Path(spec_path)
+        self.fmt = fmt or self._detect_format()
+
+        if not self.spec_path.exists():
+            raise SpecLoaderError(f"Spec file not found: {spec_path}")
+
+        if self.spec_path.suffix not in self.SUPPORTED_EXTENSIONS:
+            raise SpecLoaderError(
+                f"Unsupported file format: {self.spec_path.suffix}. "
+                f"Supported: {', '.join(self.SUPPORTED_EXTENSIONS)}"
+            )
+
+    def _detect_format(self) -> str:
+        """Detect the spec format from file extension."""
+        ext = self.spec_path.suffix.lower()
+        if ext in {".yaml", ".yml"}:
+            return "yaml"
+        elif ext == ".json":
+            return "json"
+        return "yaml"
+
+    def load(self) -> Dict[str, Any]:
+        """Load and parse the OpenAPI specification.
+
+        Returns:
+            Dictionary containing the parsed spec
+
+        Raises:
+            SpecLoaderError: If loading or parsing fails
+        """
+        try:
+            with open(self.spec_path, "r", encoding="utf-8") as f:
+                if self.fmt == "yaml":
+                    spec = yaml.safe_load(f)
+                else:
+                    spec = json.load(f)
+
+            if not isinstance(spec, dict):
+                raise SpecLoaderError("Invalid spec: root must be an object")
+
+            return spec
+
+        except yaml.YAMLError as e:
+            raise SpecLoaderError(f"YAML parsing error: {e}")
+        except json.JSONDecodeError as e:
+            raise SpecLoaderError(f"JSON parsing error: {e}")
+        except Exception as e:
+            raise SpecLoaderError(f"Failed to load spec: {e}")
+
+    def get_paths(self) -> Dict[str, Any]:
+        """Get all paths from the spec."""
+        return self.load().get("paths", {})
+
+    def get_schemas(self) -> Dict[str, Any]:
+        """Get all schemas from the spec."""
+        return self.load().get("components", {}).get("schemas", {})
+
+    def get_info(self) -> Dict[str, Any]:
+        """Get the info section from the spec."""
+        return self.load().get("info", {})
+
+    def get_server_url(self) -> Optional[str]:
+        """Get the server URL from the spec."""
+        servers = self.load().get("servers", [])
+        if servers:
+            return servers[0].get("url", "http://localhost:8080")
+        return None