JSON Schema generation¶

A simple usage of the library that, given generates a JSON Schema for inputs and outputs.

1. Parsing¶

In this sample we'll show the access from a remote public URL.

In [1]:

Copied!





from cwl_loader import load_cwl_from_location
from cwl2ogc import BaseCWLtypes2OGCConverter

workflow_id = "pattern-12"
cwl_document = load_cwl_from_location(
    "https://raw.githubusercontent.com/eoap/application-package-patterns/refs/heads/main/cwl-workflow/pattern-12.cwl"
)

workflow = None
for wf in cwl_document:
    if workflow_id == wf.id.split("#")[-1]:
        workflow = wf
        break

if workflow is not None:
    cwl_converter = BaseCWLtypes2OGCConverter(workflow)
else:
    raise ValueError(f"'#{workflow_id}' not found in input $graph")
from cwl_loader import load_cwl_from_location
from cwl2ogc import BaseCWLtypes2OGCConverter

workflow_id = "pattern-12"
cwl_document = load_cwl_from_location(
    "https://raw.githubusercontent.com/eoap/application-package-patterns/refs/heads/main/cwl-workflow/pattern-12.cwl"
)

workflow = None
for wf in cwl_document:
    if workflow_id == wf.id.split("#")[-1]:
        workflow = wf
        break

if workflow is not None:
    cwl_converter = BaseCWLtypes2OGCConverter(workflow)
else:
    raise ValueError(f"'#{workflow_id}' not found in input $graph")

---------------------------------------------------------------------------
ModuleNotFoundError                       Traceback (most recent call last)
Cell In[1], line 2
      1 from cwl_loader import load_cwl_from_location
----> 2 from cwl2ogc import BaseCWLtypes2OGCConverter
      3 
      4 workflow_id = "pattern-12"
      5 cwl_document = load_cwl_from_location(

ModuleNotFoundError: No module named 'cwl2ogc'

2. Inputs JSON Schema generation¶

Once the document is parsed, invoke the cwl2ogc APIs to convert the CWL inputs to the JSON schema:

In [2]:

Copied!

import sys

cwl_converter.dump_inputs_json_schema(stream=sys.stdout, pretty_print=True)
import sys

cwl_converter.dump_inputs_json_schema(stream=sys.stdout, pretty_print=True)

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[2], line 3
      1 import sys
      2 
----> 3 cwl_converter.dump_inputs_json_schema(stream=sys.stdout, pretty_print=True)

NameError: name 'cwl_converter' is not defined

2.1 Inputs validation¶

Schema can be used to fully validate an inputs dictionary (expecting JSON Schema validation errors in the example below):

In [3]:

Copied!





from jsonschema import Draft202012Validator
from jsonschema.exceptions import SchemaError


def validate(schema: dict, data: dict):
    try:
        validator = Draft202012Validator(schema)
        errors = validator.iter_errors(data) if validator is not None else []

        if errors:
            for error in errors:
                print(
                    f"[{'.'.join(error.schema_path)}] - #/{'/'.join(error.path)}: {error.message}"
                )
        else:
            print("No JSON Schema violations detected!")
    except SchemaError as schema_error:
        print(
            f"An error occurred while instantiating {Draft202012Validator.__class__.__name__}: {schema_error.message}"
        )
from jsonschema import Draft202012Validator
from jsonschema.exceptions import SchemaError


def validate(schema: dict, data: dict):
    try:
        validator = Draft202012Validator(schema)
        errors = validator.iter_errors(data) if validator is not None else []

        if errors:
            for error in errors:
                print(
                    f"[{'.'.join(error.schema_path)}] - #/{'/'.join(error.path)}: {error.message}"
                )
        else:
            print("No JSON Schema violations detected!")
    except SchemaError as schema_error:
        print(
            f"An error occurred while instantiating {Draft202012Validator.__class__.__name__}: {schema_error.message}"
        )

Define the inputs to be validate

In [4]:

Copied!





inputs = {
    "aoi": "-118.985,38.432,-118.183,38.938",
    "filesB": "EPSG:4326",
    "bands": ["green", "nir08"],
    "item": "https://planetarycomputer.microsoft.com/api/stac/v1/collections/landsat-c2-l2/items/LC08_L2SP_042033_20231007_02_T1",
}

validate(cwl_converter.get_inputs_json_schema(), inputs)
inputs = {
    "aoi": "-118.985,38.432,-118.183,38.938",
    "filesB": "EPSG:4326",
    "bands": ["green", "nir08"],
    "item": "https://planetarycomputer.microsoft.com/api/stac/v1/collections/landsat-c2-l2/items/LC08_L2SP_042033_20231007_02_T1",
}

validate(cwl_converter.get_inputs_json_schema(), inputs)

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[4], line 8
      4     "bands": ["green", "nir08"],
      5     "item": "https://planetarycomputer.microsoft.com/api/stac/v1/collections/landsat-c2-l2/items/LC08_L2SP_042033_20231007_02_T1",
      6 }
      7 
----> 8 validate(cwl_converter.get_inputs_json_schema(), inputs)

NameError: name 'cwl_converter' is not defined

3. Outputs JSON Schema generation¶

Users can reuse the BaseCWLtypes2OGCConverter instance to convert the CWL outputs to the JSON Schema:

In [5]:

Copied!

cwl_converter.dump_outputs_json_schema(stream=sys.stdout, pretty_print=True)
cwl_converter.dump_outputs_json_schema(stream=sys.stdout, pretty_print=True)

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[5], line 1
----> 1 cwl_converter.dump_outputs_json_schema(stream=sys.stdout, pretty_print=True)

NameError: name 'cwl_converter' is not defined

2.1 Outputs validation¶

Schema can be used to fully validate an outputs dictionary (JSON Schema validation expected to pass):

In [6]:

Copied!

outputs = {"example_out": "In girum imus nocte et consumimur igni"}

validate(cwl_converter.get_outputs_json_schema(), outputs)
outputs = {"example_out": "In girum imus nocte et consumimur igni"}

validate(cwl_converter.get_outputs_json_schema(), outputs)

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[6], line 3
      1 outputs = {"example_out": "In girum imus nocte et consumimur igni"}
      2 
----> 3 validate(cwl_converter.get_outputs_json_schema(), outputs)

NameError: name 'cwl_converter' is not defined