From efbe1b23faf21d62afa8b92735c392b9dbc7dfae Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Sun, 22 Mar 2026 21:06:26 +0000 Subject: [PATCH] Initial upload: mockapi - OpenAPI Mock Server Generator --- src/mockapi/core/spec_loader.py | 93 +++++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) create mode 100644 src/mockapi/core/spec_loader.py diff --git a/src/mockapi/core/spec_loader.py b/src/mockapi/core/spec_loader.py new file mode 100644 index 0000000..616a468 --- /dev/null +++ b/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