Ran against all yaml files under the cwl v1.2 tests directory

Slim down our schemas after creation

Author: alexiswl

cwl_utils/inputs_schema_gen.py

@@ -8,6 +8,7 @@
 import logging
 import sys
 import json
+from copy import deepcopy
 from pathlib import Path
 from typing import Any, List, Union, Dict, Optional
 from urllib.parse import urlparse
@@ -18,8 +19,7 @@
 from cwl_utils.parser import load_document_by_uri, InputArraySchemaTypes, \
     InputEnumSchemaTypes, InputRecordSchemaTypes, File, Directory, WorkflowInputParameter, InputRecordSchema, \
     InputEnumSchema, InputArraySchema, Workflow, CommandLineTool
-from cwl_utils.utils import sanitise_schema_field, is_uri, to_pascal_case, get_value_from_uri, is_local_uri, \
-    load_schema_from_uri
+from cwl_utils.utils import sanitise_schema_field, is_uri, to_pascal_case, get_value_from_uri, is_local_uri
 
 _logger = logging.getLogger("cwl-inputs-schema-gen")  # pylint: disable=invalid-name
 defaultStreamHandler = logging.StreamHandler()  # pylint: disable=invalid-name
@@ -35,6 +35,9 @@
     "string": "string",
     "int": "integer",
     "float": "number",
+    "long": "number",
+    "double": "number",
+    "null": "null"
 }
 
 JSON_TEMPLATE_PATH = Path(__file__).parent.joinpath(
@@ -123,7 +126,11 @@ def generate_type_dict_from_type(self, type_item) -> Dict:
                 return {
                     "type": PRIMITIVE_TYPES_MAPPING[type_item]
                 }
-            elif type_item in ["File", "Directory"]:
+            elif type_item in ["stdin"]:
+                return {
+                    "$ref": f"#/definitions/File"
+                }
+            elif type_item in ["File", "Directory", "Any"]:
                 return {
                     "$ref": f"#/definitions/{type_item}"
                 }
@@ -148,7 +155,7 @@ def generate_type_dict_from_type(self, type_item) -> Dict:
             return {
                 "type": "object",
                 "properties": {
-                    get_value_from_uri(prop.id): self.generate_type_dict_from_type(prop.type_)
+                    get_value_from_uri(prop.name): self.generate_type_dict_from_type(prop.type_)
                     for prop in type_item.fields
                 }
             }
@@ -162,6 +169,16 @@ def generate_type_dict_from_type(self, type_item) -> Dict:
                 }
             else:
                 raise ValueError(f"Unknown type: {type_item}")
+        elif isinstance(type_item, List):
+            # Nested schema
+            return {
+                "oneOf": list(
+                    map(
+                        lambda type_iter: self.generate_type_dict_from_type(type_iter),
+                        type_item
+                    )
+                )
+            }
         else:
             raise ValueError(f"Unknown type: {type_item}")
 
@@ -222,7 +239,7 @@ def generate_json_schema_property_from_input_parameter(input_parameter: Workflow
     return JSONSchemaProperty(
         name=input_name,
         type_=input_parameter.type_,
-        description=doc
+        description=doc if doc is not None else ""
     )
 
 
@@ -235,10 +252,15 @@ def generate_definition_from_schema(schema: InputRecordSchema) -> Dict:
 
     # Sanitise each field of the schema
     sanitised_fields = {}
-    for field_key, field_value in schema.type_.get("fields").items():
+
+    for field in schema.fields:
         sanitised_fields.update(
             {
-                field_key: sanitise_schema_field(field_value)
+                get_value_from_uri(field.name): sanitise_schema_field(
+                    {
+                        "type": field.type_
+                    }
+                )
             }
         )
 
@@ -267,13 +289,13 @@ def generate_definition_from_schema(schema: InputRecordSchema) -> Dict:
         prop = JSONSchemaProperty(
             name=prop_name,
             type_=prop_obj.get("type"),
-            description=prop_obj.get("doc"),
+            description=prop_obj.get("doc", ""),
             required=required
         )
         property_list.append(prop)
 
     return {
-        to_pascal_case(schema.type_.get('name')): {
+        to_pascal_case(get_value_from_uri(schema.name)): {
             "type": "object",
             "properties": {
                 prop.name: prop.type_dict
@@ -307,13 +329,41 @@ def cwl_to_jsonschema(cwl_obj: Union[Workflow, CommandLineTool]) -> Any:
     # Load in all $imports to be referred by complex input types
     workflow_schema_definitions_list = list(
         map(
-            lambda import_iter: generate_definition_from_schema(
-                load_schema_from_uri(import_iter)
+            lambda kv_schema_tuple_iter: generate_definition_from_schema(
+                cwl_obj.loadingOptions.idx.get(kv_schema_tuple_iter[1][0])
             ),
-            cwl_obj.loadingOptions.imports
+            filter(
+                lambda idx_iter:
+                    isinstance(idx_iter[1][0], InputRecordSchemaTypes) or
+                    isinstance(idx_iter[1][0], InputArraySchemaTypes),
+                cwl_obj.loadingOptions.idx
+            )
         )
     )
 
+    if cwl_obj.requirements is not None:
+        try:
+            schema_def_requirement = next(
+                filter(
+                    lambda requirement_iter: requirement_iter.class_ == "SchemaDefRequirement",
+                    cwl_obj.requirements
+                )
+            )
+
+            workflow_schema_definitions_list.extend(
+                list(
+                    map(
+                        lambda schema_def_iter: generate_definition_from_schema(
+                            schema_def_iter
+                        ),
+                        schema_def_requirement.types
+                    )
+                )
+            )
+
+        except StopIteration:
+            pass
+
     # Convert schema definitions to dict
     workflow_schema_definitions_dict = {}
     for schema_definition in workflow_schema_definitions_list:
@@ -334,7 +384,19 @@ def cwl_to_jsonschema(cwl_obj: Union[Workflow, CommandLineTool]) -> Any:
         {
             "type": "object",
             "properties": {
-                prop.name: prop.type_dict
+                prop.name: {
+                    "oneOf": [
+                        {
+                            "type": "null"
+                        },
+                        prop.type_dict
+                    ]
+                }
+                if prop.required is False
+                else
+                {
+                    prop.name: prop.type_dict
+                }
                 for prop in properties
             },
             "required": [
@@ -350,6 +412,80 @@ def cwl_to_jsonschema(cwl_obj: Union[Workflow, CommandLineTool]) -> Any:
         workflow_schema_definitions_dict
     )
 
+    # Slim down the schema as required
+    input_json_schema = slim_definitions(input_json_schema)
+
+    return input_json_schema
+
+
+def slim_definitions(input_json_schema: Dict) -> Dict:
+    """
+    We have quite a few definitions that we're likely not using, particularly for a simple workflow.
+
+    Traverse the properties and return all definitions that are used.
+
+    Remove all other definitions
+    """
+
+    # Traverse the properties and return all definitions that are used
+    # https://stackoverflow.com/a/77537867/6946787
+    def _recursive_search(json_data, target_key, result=None):
+        if result is None:
+            result = []
+
+        if isinstance(json_data, dict):
+            for key, value in json_data.items():
+                if key == target_key:
+                    result.append(value)
+                else:
+                    _recursive_search(value, target_key, result)
+        elif isinstance(json_data, list):
+            for item in json_data:
+                _recursive_search(item, target_key, result)
+
+        return result
+
+    # Get all the property dependencies
+    def _get_all_ref_attributes(json_object):
+        return _recursive_search(json_object, "$ref")
+
+    def get_property_dependencies(
+        property_dict: Dict,
+        input_json_schema: Dict,
+        existing_property_dependencies: List = None
+    ) -> List[str]:
+        # Initialise return list
+        if existing_property_dependencies is None:
+            existing_property_dependencies = []
+
+        # All reference attributes
+        for reference_attribute in _get_all_ref_attributes(property_dict):
+            # Get the value from the reference attribute
+            reference_value = get_value_from_uri(reference_attribute)
+            # If the reference value is not in the existing property dependencies, add it
+            if reference_value not in existing_property_dependencies:
+                existing_property_dependencies.append(reference_value)
+                # Get the property dependencies of the reference value
+                existing_property_dependencies.extend(
+                    get_property_dependencies(
+                        input_json_schema['definitions'][reference_value],
+                        input_json_schema,
+                        existing_property_dependencies
+                    )
+                )
+
+        return existing_property_dependencies
+
+    # Copy schema
+    input_json_schema = deepcopy(input_json_schema)
+
+    # Get required definitions
+    required_definitions = get_property_dependencies(input_json_schema.get("properties"), input_json_schema)
+
+    for definition_key in list(input_json_schema['definitions'].keys()):
+        if definition_key not in required_definitions:
+            del input_json_schema['definitions'][definition_key]
+
     return input_json_schema
 
 
@@ -430,7 +566,7 @@ def run(args: argparse.Namespace) -> int:
     except Exception as e:
         _logger.exception("Failed to generate JSON Schema from CWL inputs object. Error: %s", e)
         return 1
-    args.output.write(json.dumps(jsonschema, indent=2))
+    args.output.write(json.dumps(jsonschema, indent=2) + "\n")
 
     return 0