json-to-openapi/json_to_openapi/schema_generator.py

"""OpenAPI schema generation module."""

from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
import json
import yaml

from json_to_openapi.analyzer import JsonAnalyzer, TypeInfo, parse_json_file


@dataclass
class EndpointInfo:
    """Information about an API endpoint."""
    path: str = "/"
    method: str = "get"
    summary: str = ""
    description: str = ""
    tags: List[str] = field(default_factory=list)
    operation_id: str = ""


class OpenAPIGenerator:
    """Generates OpenAPI 3.0 specifications from JSON data."""

    def __init__(self, title: str = "API", version: str = "1.0.0"):
        self.title = title
        self.version = version
        self.analyzer = JsonAnalyzer()

    def generate(
        self,
        data: Any,
        endpoint: Optional[EndpointInfo] = None,
        description: str = ""
    ) -> Dict[str, Any]:
        """Generate an OpenAPI specification from JSON data."""
        type_info = self.analyzer.analyze(data)
        schema = self._type_info_to_schema(type_info)

        spec: Dict[str, Any] = {
            "openapi": "3.0.3",
            "info": {
                "title": self.title,
                "version": self.version,
            },
            "paths": {},
        }

        if description:
            spec["info"]["description"] = description

        path = endpoint.path if endpoint else "/"
        method = endpoint.method.lower() if endpoint else "get"

        method_spec: Dict[str, Any] = {
            "summary": endpoint.summary if endpoint else f"Get {self.title}",
            "description": endpoint.description if endpoint else description,
            "operationId": endpoint.operation_id if endpoint else f"get{self.title.replace(' ', '')}",
            "responses": {
                "200": {
                    "description": "Successful response",
                    "content": {
                        "application/json": {
                            "schema": schema
                        }
                    }
                }
            }
        }

        if endpoint and endpoint.tags:
            method_spec["tags"] = endpoint.tags

        spec["paths"] = {
            path: {
                method: method_spec
            }
        }

        return spec

    def _type_info_to_schema(self, type_info: TypeInfo) -> Dict[str, Any]:
        schema: Dict[str, Any] = {}

        schema["type"] = type_info.type_name

        if type_info.format:
            schema["format"] = type_info.format

        if type_info.nullable:
            schema["nullable"] = True

        if type_info.enum_values:
            schema["enum"] = type_info.enum_values

        if type_info.type_name == "object" and type_info.properties:
            properties: Dict[str, Any] = {}
            required_fields: List[str] = []

            for prop_name, prop_type in type_info.properties.items():
                properties[prop_name] = self._type_info_to_schema(prop_type)
                required_fields.append(prop_name)

            schema["properties"] = properties
            if required_fields:
                schema["required"] = required_fields

        elif type_info.type_name == "array" and type_info.items:
            schema["items"] = self._type_info_to_schema(type_info.items)

        return schema

    def generate_batch(
        self,
        files: List[str],
        output_dir: Optional[str] = None,
        combined: bool = False
    ) -> Dict[str, Any]:
        """Process multiple JSON files and generate OpenAPI specs."""
        results: List[Dict[str, Any]] = []
        combined_spec: Optional[Dict[str, Any]] = None

        if combined:
            combined_spec = {
                "openapi": "3.0.3",
                "info": {
                    "title": self.title,
                    "version": self.version,
                },
                "paths": {},
            }

        for file_path in files:
            data = parse_json_file(file_path)
            spec = self.generate(data)
            results.append({
                "file": file_path,
                "spec": spec
            })

            if combined_spec and "paths" in spec:
                for path, methods in spec["paths"].items():
                    if path not in combined_spec["paths"]:
                        combined_spec["paths"][path] = {}
                    combined_spec["paths"][path].update(methods)

        if combined and combined_spec:
            return {
                "combined": combined_spec,
                "individual": results
            }

        return {"individual": results}

    def to_yaml(self, spec: Dict[str, Any]) -> str:
        """Convert OpenAPI spec to YAML string."""
        return yaml.dump(spec, default_flow_style=False, sort_keys=False)

    def to_json(self, spec: Dict[str, Any]) -> str:
        """Convert OpenAPI spec to JSON string."""
        return json.dumps(spec, indent=2)

    def save_spec(
        self,
        spec: Dict[str, Any],
        output_path: str,
        format: str = "yaml"
    ) -> None:
        """Save OpenAPI spec to a file."""
        if format.lower() == "yaml":
            content = self.to_yaml(spec)
            if not output_path.endswith((".yaml", ".yml")):
                output_path += ".yaml"
        else:
            content = self.to_json(spec)
            if not output_path.endswith(".json"):
                output_path += ".json"

        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(content)