Initial upload with comprehensive README and tests
This commit is contained in:
180
json_to_openapi/schema_generator.py
Normal file
180
json_to_openapi/schema_generator.py
Normal file
@@ -0,0 +1,180 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user