Add structured CLI and output schemas

Author: jmchilton

planemo/cli.py

@@ -32,6 +32,10 @@
     "t": "test",
     "s": "serve",
 }
+INTERNAL_COMMANDS = [
+    "create_gist",
+    "shed_download",
+]
 
 
 class PlanemoCliContext(PlanemoContext):
@@ -186,6 +190,7 @@ def planemo(ctx, config, directory, verbose, configure_logging=True):
 
 __all__ = (
     "command_function",
+    "INTERNAL_COMMANDS",
     "list_cmds",
     "name_to_command",
     "planemo",

planemo/cli_metadata.py

@@ -0,0 +1,125 @@
+"""Structured metadata for Planemo's Click command tree."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+import click
+
+from planemo import __version__
+from planemo.cli import (
+    COMMAND_ALIASES,
+    INTERNAL_COMMANDS,
+    list_cmds,
+    name_to_command,
+)
+
+SCHEMA_VERSION = "0.1"
+
+
+def iter_command_names(include_internal: bool = False) -> list[str]:
+    """Return command names included in public CLI metadata."""
+    commands = list_cmds()
+    if not include_internal:
+        commands = [command for command in commands if command not in INTERNAL_COMMANDS]
+    return commands
+
+
+def load_command_metadata(command_name: str) -> dict[str, Any]:
+    """Load metadata for one Planemo command."""
+    return serialize_click_command(command_name, name_to_command(command_name))
+
+
+def load_planemo_metadata(include_internal: bool = False) -> dict[str, Any]:
+    """Load metadata for the Planemo command line interface."""
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "program": "planemo",
+        "planemo_version": __version__,
+        "commands": [load_command_metadata(command_name) for command_name in iter_command_names(include_internal)],
+        "aliases": dict(sorted(COMMAND_ALIASES.items())),
+    }
+
+
+def serialize_click_command(command_name: str, command: click.Command) -> dict[str, Any]:
+    """Serialize one Click command into JSON-compatible metadata."""
+    context = click.Context(command, info_name=command_name)
+    return {
+        "name": command_name,
+        "module": f"planemo.commands.cmd_{command_name}",
+        "help": command.help,
+        "short_help": command.short_help,
+        "usage": command.get_usage(context).removeprefix("Usage: "),
+        "internal": command_name in INTERNAL_COMMANDS,
+        "hidden": getattr(command, "hidden", False),
+        "params": [serialize_click_param(param) for param in command.params],
+    }
+
+
+def serialize_click_param(param: click.Parameter) -> dict[str, Any]:
+    """Serialize one Click parameter into JSON-compatible metadata."""
+    planemo_config = getattr(param, "planemo_config", {})
+    metadata = {
+        "kind": "option" if isinstance(param, click.Option) else "argument",
+        "name": param.name,
+        "opts": list(getattr(param, "opts", [])),
+        "secondary_opts": list(getattr(param, "secondary_opts", [])),
+        "help": getattr(param, "help", None),
+        "required": param.required,
+        "human_readable_name": param.human_readable_name,
+        "multiple": param.multiple,
+        "nargs": param.nargs,
+        "type": serialize_click_type(param.type),
+        "default": _json_value(planemo_config.get("declared_default", getattr(param, "default", None))),
+        "is_flag": getattr(param, "is_flag", False),
+        "flag_value": _json_value(getattr(param, "flag_value", None)),
+        "envvar": _json_value(getattr(param, "envvar", None)),
+        "hidden": getattr(param, "hidden", False),
+        "prompt": getattr(param, "prompt", None),
+        "planemo_config": _json_value(planemo_config),
+    }
+    if isinstance(param, click.Option):
+        metadata["is_bool_flag"] = param.is_bool_flag
+    return metadata
+
+
+def serialize_click_type(param_type: click.ParamType) -> dict[str, Any]:
+    """Serialize Click parameter type details where Click exposes them."""
+    metadata: dict[str, Any] = {"name": param_type.name}
+    if isinstance(param_type, click.Choice):
+        metadata["choices"] = list(param_type.choices)
+        metadata["case_sensitive"] = param_type.case_sensitive
+    elif isinstance(param_type, click.IntRange):
+        metadata.update(
+            {
+                "min": param_type.min,
+                "max": param_type.max,
+                "min_open": param_type.min_open,
+                "max_open": param_type.max_open,
+            }
+        )
+    elif isinstance(param_type, click.FloatRange):
+        metadata.update(
+            {
+                "min": param_type.min,
+                "max": param_type.max,
+                "min_open": param_type.min_open,
+                "max_open": param_type.max_open,
+            }
+        )
+    elif isinstance(param_type, click.Path):
+        for attr in ("exists", "file_okay", "dir_okay", "writable", "readable", "resolve_path", "allow_dash"):
+            if hasattr(param_type, attr):
+                metadata[attr] = getattr(param_type, attr)
+    return _json_value(metadata)
+
+
+def _json_value(value: Any) -> Any:
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    if isinstance(value, Mapping):
+        return {str(key): _json_value(nested_value) for key, nested_value in value.items()}
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
+        return [_json_value(nested_value) for nested_value in value]
+    return repr(value)

planemo/commands/cmd_cli_metadata.py

@@ -0,0 +1,23 @@
+"""Module describing the planemo ``cli_metadata`` command."""
+
+import json
+
+import click
+
+from planemo.cli_metadata import (
+    load_command_metadata,
+    load_planemo_metadata,
+)
+
+
+@click.command("cli_metadata")
+@click.option("--format", "output_format", default="json", type=click.Choice(["json"]), show_default=True)
+@click.option("--command", "command_name", help="Only export metadata for the selected command.")
+@click.option("--include-internal", is_flag=True, help="Include internal commands not documented as public API.")
+def cli(output_format, command_name, include_internal):
+    """Export structured metadata for Planemo CLI commands."""
+    if command_name:
+        metadata = load_command_metadata(command_name)
+    else:
+        metadata = load_planemo_metadata(include_internal=include_internal)
+    click.echo(json.dumps(metadata, indent=2, sort_keys=True))

planemo/commands/cmd_database_list.py

@@ -7,7 +7,7 @@
 from planemo.database import create_database_source
 
 
-@click.command("database_create")
+@click.command("database_list")
 @options.profile_database_options()
 @options.docker_config_options()
 @command_function

planemo/commands/cmd_output_schema.py

@@ -0,0 +1,18 @@
+"""Module describing the planemo ``output_schema`` command."""
+
+import json
+
+import click
+
+from planemo.output_schemas import (
+    SCHEMA_MODELS,
+    load_output_schemas,
+)
+
+
+@click.command("output_schema")
+@click.option("--format", "output_format", default="json", type=click.Choice(["json"]), show_default=True)
+@click.option("--schema", "schema_name", type=click.Choice(sorted(SCHEMA_MODELS)), help="Only export one schema.")
+def cli(output_format, schema_name):
+    """Export JSON Schemas for Planemo machine-readable outputs."""
+    click.echo(json.dumps(load_output_schemas(schema_name), indent=2, sort_keys=True))

planemo/config.py

@@ -124,8 +124,22 @@ def callback(ctx, param, value):
         assert name
         kwargs["envvar"] = f"PLANEMO_{name.upper()}"
 
+    planemo_config = {
+        "use_global_config": use_global_config,
+        "extra_global_config_vars": extra_global_config_vars,
+        "use_env_var": use_env_var,
+    }
+    if default_specified:
+        planemo_config["declared_default"] = default
+
     option = click.option(*args, **kwargs)
-    return option
+
+    def decorator(f):
+        f = option(f)
+        f.__click_params__[-1].planemo_config = planemo_config
+        return f
+
+    return decorator
 
 
 def global_config_path(config_path: Optional[str] = None) -> str:

planemo/output_models.py

@@ -0,0 +1,19 @@
+"""Pydantic models for Planemo machine-readable outputs."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import RootModel
+
+from planemo.test.models import PlanemoTestReport
+
+
+class PlanemoRunOutputs(RootModel[dict[str, Any]]):
+    """Permissive model for ``planemo run --output_json`` outputs."""
+
+
+__all__ = (
+    "PlanemoRunOutputs",
+    "PlanemoTestReport",
+)

planemo/output_schemas.py

@@ -0,0 +1,42 @@
+"""JSON Schema exports for Planemo machine-readable outputs."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from planemo import __version__
+from planemo.output_models import (
+    PlanemoRunOutputs,
+    PlanemoTestReport,
+)
+
+SCHEMA_VERSION = "0.1"
+
+SCHEMA_MODELS = {
+    "run-outputs": PlanemoRunOutputs,
+    "test-report": PlanemoTestReport,
+}
+
+
+def load_output_schemas(schema_name: str | None = None) -> dict[str, Any]:
+    """Return Planemo output JSON Schemas."""
+    schemas = {}
+    for name, model in SCHEMA_MODELS.items():
+        if schema_name is None or name == schema_name:
+            schema = model.model_json_schema()
+            schema["$schema"] = "https://json-schema.org/draft/2020-12/schema"
+            schemas[name] = schema
+    if schema_name is not None and schema_name not in schemas:
+        raise KeyError(schema_name)
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "planemo_version": __version__,
+        "schemas": schemas,
+    }
+
+
+__all__ = (
+    "SCHEMA_MODELS",
+    "SCHEMA_VERSION",
+    "load_output_schemas",
+)

planemo/test/models.py

@@ -0,0 +1,81 @@
+"""Pydantic models for Planemo machine-readable test outputs."""
+
+from __future__ import annotations
+
+from typing import (
+    Any,
+    Literal,
+)
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    field_validator,
+    model_validator,
+)
+
+PlanemoTestStatus = Literal["success", "failure", "error", "skip"]
+
+
+class PlanemoTestSummary(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    num_tests: int
+    num_failures: int
+    num_skips: int
+    num_errors: int
+
+
+class PlanemoTestCaseData(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    status: PlanemoTestStatus | None = None
+    inputs: dict[str, Any] | None = None
+    job: dict[str, Any] | None = None
+    invocation_details: dict[str, Any] | None = None
+    problem_log: str | None = None
+    output_problems: list[str] | None = None
+    execution_problem: str | None = None
+    start_datetime: str | None = None
+    end_datetime: str | None = None
+
+    @field_validator("status", mode="before")
+    @classmethod
+    def normalize_legacy_status(cls, value):
+        if value == "skipped":
+            return "skip"
+        return value
+
+
+class PlanemoTestCase(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    id: str
+    has_data: bool
+    data: PlanemoTestCaseData | None = None
+    doc: str | None = None
+    test_type: str | None = None
+
+    @model_validator(mode="after")
+    def require_data_when_present(self):
+        if self.has_data and self.data is None:
+            raise ValueError("data is required when has_data is true")
+        return self
+
+
+class PlanemoTestReport(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    version: str = "0.1"
+    tests: list[PlanemoTestCase]
+    summary: PlanemoTestSummary | None = None
+    exit_code: int | None = None
+
+
+__all__ = (
+    "PlanemoTestCase",
+    "PlanemoTestCaseData",
+    "PlanemoTestReport",
+    "PlanemoTestStatus",
+    "PlanemoTestSummary",
+)

requirements.txt

@@ -15,6 +15,7 @@ lxml
 oyaml
 packaging
 pathvalidate
+pydantic>=2
 pyyaml
 ruamel.yaml
 virtualenv