Merge pull request #175 from jmchilton/app_models

CLI Tweaks

Author: jmchilton

gxformat2/converter.py

@@ -98,22 +98,28 @@ def main(argv=None):
     args = _parser().parse_args(argv)
 
     format2_path = args.input_path
-    output_path = args.output_path or (format2_path + ".gxwf.yml")
+    output_path = args.output or args.output_path
 
-    workflow_directory = os.path.abspath(format2_path)
+    workflow_directory = os.path.dirname(os.path.abspath(format2_path))
 
     with open(format2_path) as f:
         has_workflow = ordered_load(f)
 
     output = python_to_workflow(has_workflow, workflow_directory=workflow_directory)
-    with open(output_path, "w") as f:
-        json.dump(output, f, indent=4)
+    output_text = json.dumps(output, indent=4) + "\n"
+
+    if output_path:
+        with open(output_path, "w") as f:
+            f.write(output_text)
+    else:
+        sys.stdout.write(output_text)
 
 
 def _parser():
     parser = argparse.ArgumentParser(description=SCRIPT_DESCRIPTION)
     parser.add_argument("input_path", metavar="INPUT", type=str, help="input workflow path (.gxwf.yml)")
     parser.add_argument("output_path", metavar="OUTPUT", type=str, nargs="?", help="output workflow path (.ga)")
+    parser.add_argument("--output", "-o", help="output file (default: stdout)")
     return parser
 
 

gxformat2/export.py

@@ -5,6 +5,7 @@
 """
 
 import argparse
+import io
 import json
 import sys
 from collections import OrderedDict
@@ -138,20 +139,33 @@ def main(argv=None):
     args = _parser().parse_args(argv)
 
     format2_path = args.input_path
-    output_path = args.output_path or (format2_path + ".gxwf.yml")
+    output_path = args.output or args.output_path
     with open(format2_path) as f:
         native_workflow_dict = json.load(f)
 
     as_dict = from_galaxy_native(native_workflow_dict, compact=args.compact)
-    with open(output_path, "w") as f:
-        ordered_dump(as_dict, f)
+
+    if args.json_output:
+        output_text = json.dumps(as_dict, indent=4) + "\n"
+    else:
+        stream = io.StringIO()
+        ordered_dump(as_dict, stream)
+        output_text = stream.getvalue()
+
+    if output_path:
+        with open(output_path, "w") as f:
+            f.write(output_text)
+    else:
+        sys.stdout.write(output_text)
 
 
 def _parser():
     parser = argparse.ArgumentParser(description=SCRIPT_DESCRIPTION)
     parser.add_argument("input_path", metavar="INPUT", type=str, help="input workflow path (.ga)")
-    parser.add_argument("output_path", metavar="OUTPUT", type=str, nargs="?", help="output workflow path (.gxfw.yml)")
-    parser.add_argument("--compact", action="store_true", help="Generate compact workflow without position information")
+    parser.add_argument("output_path", metavar="OUTPUT", type=str, nargs="?", help="output workflow path (.gxwf.yml)")
+    parser.add_argument("--output", "-o", help="output file (default: stdout)")
+    parser.add_argument("--compact", action="store_true", help="generate compact workflow without position information")
+    parser.add_argument("--json", dest="json_output", action="store_true", help="output JSON instead of YAML")
     return parser
 
 

gxformat2/schema/json_schema.py

@@ -0,0 +1,53 @@
+"""JSON Schema export for gxformat2 workflow models.
+
+Provides export functions that generate standard Draft 2020-12 JSON Schema
+from gxformat2's Pydantic models. Exported schemas are intended for consumption
+by external validators (Python jsonschema, TypeScript ajv, VSCode YAML extension).
+"""
+
+import json
+from typing import (
+    Any,
+    Dict,
+    Type,
+)
+
+from pydantic import BaseModel
+from pydantic.json_schema import GenerateJsonSchema
+from typing_extensions import Literal
+
+MODE = Literal["validation", "serialization"]
+
+
+class GxFormat2GenerateJsonSchema(GenerateJsonSchema):
+    """Custom JSON Schema generator that injects $schema dialect."""
+
+    def generate(self, schema, mode: MODE = "validation"):
+        """Generate JSON Schema with $schema dialect injected."""
+        json_schema = super().generate(schema, mode=mode)
+        json_schema["$schema"] = self.schema_dialect
+        return json_schema
+
+
+def workflow_json_schema(*, strict: bool = False, mode: MODE = "validation") -> Dict[str, Any]:
+    """Export GalaxyWorkflow JSON Schema.
+
+    Args:
+        strict: If True, use strict model (extra="forbid" — rejects unknown keys).
+        mode: Pydantic schema mode ("validation" or "serialization").
+    """
+    model_class: Type[BaseModel]
+    if strict:
+        from .gxformat2_strict import GalaxyWorkflow as model_class
+    else:
+        from .gxformat2 import GalaxyWorkflow as model_class
+
+    return model_class.model_json_schema(
+        schema_generator=GxFormat2GenerateJsonSchema,
+        mode=mode,
+    )
+
+
+def workflow_json_schema_string(*, strict: bool = False, mode: MODE = "validation") -> str:
+    """Export GalaxyWorkflow JSON Schema as formatted string."""
+    return json.dumps(workflow_json_schema(strict=strict, mode=mode), indent=4)