Initial upload with comprehensive README and tests
This commit is contained in:
265
json_to_openapi/cli.py
Normal file
265
json_to_openapi/cli.py
Normal file
@@ -0,0 +1,265 @@
|
||||
"""CLI interface for JSON to OpenAPI Generator."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import glob
|
||||
from typing import List, Optional
|
||||
from dataclasses import dataclass
|
||||
|
||||
import click
|
||||
from json_to_openapi.analyzer import parse_json_file
|
||||
from json_to_openapi.schema_generator import OpenAPIGenerator, EndpointInfo
|
||||
|
||||
|
||||
@dataclass
|
||||
class CLIContext:
|
||||
"""CLI context object."""
|
||||
verbose: bool = False
|
||||
quiet: bool = False
|
||||
|
||||
|
||||
def echo(msg: str, ctx: CLIContext) -> None:
|
||||
"""Print message if not in quiet mode."""
|
||||
if not ctx.quiet:
|
||||
click.echo(msg)
|
||||
|
||||
|
||||
def debug(msg: str, ctx: CLIContext) -> None:
|
||||
"""Print debug message if in verbose mode."""
|
||||
if ctx.verbose:
|
||||
click.echo(f"[DEBUG] {msg}")
|
||||
|
||||
|
||||
@click.group()
|
||||
@click.option("-v", "--verbose", is_flag=True, help="Enable verbose output")
|
||||
@click.option("-q", "--quiet", is_flag=True, help="Suppress all output except errors")
|
||||
@click.pass_context
|
||||
def main(ctx: click.Context, verbose: bool, quiet: bool) -> None:
|
||||
"""JSON to OpenAPI Generator - Convert JSON files to OpenAPI 3.0 specifications."""
|
||||
ctx.ensure_object(CLIContext)
|
||||
ctx.obj.verbose = verbose
|
||||
ctx.obj.quiet = quiet
|
||||
|
||||
|
||||
@main.command("convert")
|
||||
@click.argument("input_file", type=click.Path(exists=True))
|
||||
@click.option("-o", "--output", type=click.Path(), help="Output file path")
|
||||
@click.option("-f", "--format", type=click.Choice(["yaml", "json"]), default="yaml",
|
||||
help="Output format (default: yaml)")
|
||||
@click.option("-t", "--title", default="My API", help="API title")
|
||||
@click.option("-v", "--version", default="1.0.0", help="API version")
|
||||
@click.option("-d", "--description", default="", help="API description")
|
||||
@click.pass_context
|
||||
def convert(
|
||||
ctx: click.Context,
|
||||
input_file: str,
|
||||
output: Optional[str],
|
||||
format: str,
|
||||
title: str,
|
||||
version: str,
|
||||
description: str
|
||||
) -> None:
|
||||
"""Convert a single JSON file to OpenAPI specification."""
|
||||
cli_ctx = ctx.obj
|
||||
debug(f"Processing file: {input_file}", cli_ctx)
|
||||
|
||||
try:
|
||||
data = parse_json_file(input_file)
|
||||
debug("Parsed JSON successfully", cli_ctx)
|
||||
|
||||
generator = OpenAPIGenerator(title=title, version=version)
|
||||
spec = generator.generate(data, description=description)
|
||||
debug("Generated OpenAPI spec", cli_ctx)
|
||||
|
||||
if not output:
|
||||
base_name = os.path.splitext(input_file)[0]
|
||||
output = f"{base_name}.{format}"
|
||||
|
||||
generator.save_spec(spec, output, format)
|
||||
echo(f"Generated: {output}", cli_ctx)
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@main.command("interactive")
|
||||
@click.option("-t", "--title", default="My API", help="API title")
|
||||
@click.option("-v", "--version", default="1.0.0", help="API version")
|
||||
@click.pass_context
|
||||
def interactive(ctx: click.Context, title: str, version: str) -> None:
|
||||
"""Interactive mode for customizing OpenAPI specification."""
|
||||
cli_ctx = ctx.obj
|
||||
|
||||
echo("JSON to OpenAPI Generator - Interactive Mode", cli_ctx)
|
||||
echo("=" * 50, cli_ctx)
|
||||
|
||||
input_file = click.prompt("Enter path to JSON file", type=click.Path(exists=True))
|
||||
|
||||
try:
|
||||
data = parse_json_file(input_file)
|
||||
|
||||
endpoint_path = click.prompt("Endpoint path", default="/items", type=str)
|
||||
endpoint_method = click.prompt("HTTP method", default="get",
|
||||
type=click.Choice(["get", "post", "put", "delete", "patch"]))
|
||||
endpoint_summary = click.prompt("Endpoint summary", default=f"Get {title} items", type=str)
|
||||
endpoint_description = click.prompt("Endpoint description", default="", type=str)
|
||||
|
||||
add_tags = click.confirm("Add tags to endpoint?", default=False)
|
||||
tags: List[str] = []
|
||||
if add_tags:
|
||||
tag_input = click.prompt("Enter tags (comma-separated)", type=str)
|
||||
tags = [t.strip() for t in tag_input.split(",") if t.strip()]
|
||||
|
||||
output_format = click.prompt("Output format", default="yaml",
|
||||
type=click.Choice(["yaml", "json"]))
|
||||
output_path = click.prompt("Output file path", type=str)
|
||||
|
||||
endpoint = EndpointInfo(
|
||||
path=endpoint_path,
|
||||
method=endpoint_method,
|
||||
summary=endpoint_summary,
|
||||
description=endpoint_description,
|
||||
tags=tags
|
||||
)
|
||||
|
||||
generator = OpenAPIGenerator(title=title, version=version)
|
||||
spec = generator.generate(data, endpoint=endpoint)
|
||||
|
||||
generator.save_spec(spec, output_path, output_format)
|
||||
echo(f"\nGenerated: {output_path}", cli_ctx)
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@main.command("batch")
|
||||
@click.argument("input_pattern", type=str)
|
||||
@click.option("-o", "--output-dir", type=click.Path(exists=False), help="Output directory")
|
||||
@click.option("-f", "--format", type=click.Choice(["yaml", "json"]), default="yaml",
|
||||
help="Output format (default: yaml)")
|
||||
@click.option("-t", "--title", default="My API", help="API title")
|
||||
@click.option("-v", "--version", default="1.0.0", help="API version")
|
||||
@click.option("-c", "--combined", is_flag=True, help="Generate combined OpenAPI spec")
|
||||
@click.option("-p", "--parallel", is_flag=True, help="Process files in parallel")
|
||||
@click.pass_context
|
||||
def batch(
|
||||
ctx: click.Context,
|
||||
input_pattern: str,
|
||||
output_dir: Optional[str],
|
||||
format: str,
|
||||
title: str,
|
||||
version: str,
|
||||
combined: bool,
|
||||
parallel: bool
|
||||
) -> None:
|
||||
"""Process multiple JSON files matching a pattern."""
|
||||
cli_ctx = ctx.obj
|
||||
|
||||
files = glob.glob(input_pattern)
|
||||
if not files:
|
||||
echo(f"No files found matching: {input_pattern}", cli_ctx)
|
||||
sys.exit(1)
|
||||
|
||||
echo(f"Found {len(files)} files to process", cli_ctx)
|
||||
|
||||
if output_dir and not os.path.exists(output_dir):
|
||||
os.makedirs(output_dir)
|
||||
debug(f"Created output directory: {output_dir}", cli_ctx)
|
||||
|
||||
try:
|
||||
generator = OpenAPIGenerator(title=title, version=version)
|
||||
|
||||
if combined:
|
||||
result = generator.generate_batch(files, output_dir, combined=True)
|
||||
combined_spec = result["combined"]
|
||||
|
||||
output_path = os.path.join(output_dir or ".", f"combined.{format}")
|
||||
generator.save_spec(combined_spec, output_path, format)
|
||||
echo(f"Generated combined spec: {output_path}", cli_ctx)
|
||||
|
||||
for item in result["individual"]:
|
||||
debug(f"Processed: {item['file']}", cli_ctx)
|
||||
else:
|
||||
for file_path in files:
|
||||
data = parse_json_file(file_path)
|
||||
spec = generator.generate(data)
|
||||
|
||||
base_name = os.path.splitext(os.path.basename(file_path))[0]
|
||||
output_path = os.path.join(output_dir or ".", f"{base_name}.{format}")
|
||||
|
||||
generator.save_spec(spec, output_path, format)
|
||||
echo(f"Generated: {output_path}", cli_ctx)
|
||||
|
||||
echo(f"Successfully processed {len(files)} files", cli_ctx)
|
||||
|
||||
except Exception as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@main.command("validate")
|
||||
@click.argument("spec_file", type=click.Path(exists=True))
|
||||
@click.pass_context
|
||||
def validate(ctx: click.Context, spec_file: str) -> None:
|
||||
"""Validate an OpenAPI specification file."""
|
||||
cli_ctx = ctx.obj
|
||||
|
||||
try:
|
||||
import yaml
|
||||
|
||||
with open(spec_file, 'r', encoding='utf-8') as f:
|
||||
spec = yaml.safe_load(f)
|
||||
|
||||
errors: List[str] = []
|
||||
|
||||
if not isinstance(spec, dict):
|
||||
errors.append("Spec must be a dictionary")
|
||||
|
||||
if "openapi" not in spec:
|
||||
errors.append("Missing 'openapi' field")
|
||||
elif spec["openapi"] != "3.0.3":
|
||||
errors.append(f"Unsupported OpenAPI version: {spec.get('openapi')}")
|
||||
|
||||
if "info" not in spec:
|
||||
errors.append("Missing 'info' field")
|
||||
else:
|
||||
info = spec["info"]
|
||||
if "title" not in info:
|
||||
errors.append("Missing 'info.title'")
|
||||
if "version" not in info:
|
||||
errors.append("Missing 'info.version'")
|
||||
|
||||
if "paths" not in spec:
|
||||
errors.append("Missing 'paths' field")
|
||||
|
||||
if errors:
|
||||
echo("Validation failed:", cli_ctx)
|
||||
for error in errors:
|
||||
echo(f" - {error}", cli_ctx)
|
||||
sys.exit(1)
|
||||
else:
|
||||
echo("Validation passed! Spec is valid OpenAPI 3.0.3", cli_ctx)
|
||||
|
||||
except yaml.YAMLError as e:
|
||||
click.echo(f"YAML error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
click.echo(f"Error: {e}", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@main.command("info")
|
||||
@click.pass_context
|
||||
def info(ctx: click.Context) -> None:
|
||||
"""Display information about the tool."""
|
||||
cli_ctx = ctx.obj
|
||||
echo("JSON to OpenAPI Generator v1.0.0", cli_ctx)
|
||||
echo("", cli_ctx)
|
||||
echo("Commands:", cli_ctx)
|
||||
echo(" convert Convert a JSON file to OpenAPI spec", cli_ctx)
|
||||
echo(" interactive Interactive mode for customization", cli_ctx)
|
||||
echo(" batch Process multiple JSON files", cli_ctx)
|
||||
echo(" validate Validate an OpenAPI spec file", cli_ctx)
|
||||
echo(" info Display this help information", cli_ctx)
|
||||
Reference in New Issue
Block a user