opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-10-29 17:42:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Enrich tool parameter JSON Schema with Galaxy metadata via model methods

Browse Source 
Add field_kwargs() methods to the pydantic parameter model hierarchy to
expose label, help, argument, options, extensions, and other Galaxy
metadata in generated JSON Schema. Each model class owns its own schema
metadata rather than relying on a central hasattr/getattr function.

Add _common_param_kwargs helper to extract label/help from InputSource
and pass them to all parameter model constructors in the factory.
This commit is contained in:
mvdbeek
2026-03-09 12:46:43 +01:00
parent 25e6cae50d
commit 6699a3dfe8
2 changed files with 165 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/tool_util/parameters/factory.py
+38
View File
@@ -8,6 +8,8 @@ from typing import (
Union,
)
from typing_extensions import TypedDict
from galaxy.tool_util.parser.interface import (
InputSource,
PageSource,
@@ -76,6 +78,23 @@ def get_color_value(input_source: InputSource) -> str:
return input_source.get("value", "#000000")
class _CommonParamKwargs(TypedDict, total=False):
label: str
help: str
def _common_param_kwargs(input_source: InputSource) -> _CommonParamKwargs:
"""Extract common metadata (label, help) from InputSource for parameter models."""
kwargs = _CommonParamKwargs()
label = input_source.parse_label()
if label:
kwargs["label"] = label
help_text = input_source.parse_help()
if help_text:
kwargs["help"] = help_text
return kwargs
def _from_input_source_galaxy(input_source: InputSource, profile: float) -> ToolParameterT:
input_type = input_source.parse_input_type()
if input_type == "param":
@@ -110,6 +129,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
min=min_int,
max=max_int,
validators=int_validators,
**_common_param_kwargs(input_source),
)
elif param_type == "boolean":
nullable = input_source.parse_optional()
@@ -119,6 +139,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
name=input_source.parse_name(),
optional=nullable,
value=value,
**_common_param_kwargs(input_source),
)
elif param_type == "text":
optional, optionality_inferred = text_input_is_optional(input_source)
@@ -131,6 +152,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
optional=optional,
validators=text_validators,
value=default_value,
**_common_param_kwargs(input_source),
)
elif param_type == "float":
optional = input_source.parse_optional()
@@ -162,6 +184,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
min=min_float,
max=max_float,
validators=float_validators,
**_common_param_kwargs(input_source),
)
elif param_type == "hidden":
optional = input_source.parse_optional()
@@ -173,6 +196,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
optional=optional,
value=value,
validators=hidden_validators,
**_common_param_kwargs(input_source),
)
elif param_type == "color":
optional = input_source.parse_optional()
@@ -181,11 +205,13 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
name=input_source.parse_name(),
optional=optional,
value=get_color_value(input_source),
**_common_param_kwargs(input_source),
)
elif param_type == "rules":
return RulesParameterModel(
type="rules",
name=input_source.parse_name(),
**_common_param_kwargs(input_source),
)
elif param_type == "data":
optional = input_source.parse_optional()
@@ -195,6 +221,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
name=input_source.parse_name(),
optional=optional,
multiple=multiple,
**_common_param_kwargs(input_source),
)
elif param_type == "data_collection":
optional = input_source.parse_optional()
@@ -205,6 +232,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
name=input_source.parse_name(),
optional=optional,
value=default_value,
**_common_param_kwargs(input_source),
)
elif param_type == "select":
# Function... example in devteam cummeRbund.
@@ -232,6 +260,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
options=options,
multiple=multiple,
validators=select_validators,
**_common_param_kwargs(input_source),
)
elif param_type == "drill_down":
multiple = input_source.get_bool("multiple", False)
@@ -246,6 +275,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
multiple=multiple,
hierarchy=hierarchy,
options=static_options,
**_common_param_kwargs(input_source),
)
elif param_type == "data_column":
multiple = input_source.get_bool("multiple", False)
@@ -269,6 +299,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
multiple=multiple,
optional=optional,
value=value,
**_common_param_kwargs(input_source),
)
elif param_type == "group_tag":
multiple = input_source.get_bool("multiple", False)
@@ -278,11 +309,13 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
name=input_source.parse_name(),
optional=optional,
multiple=multiple,
**_common_param_kwargs(input_source),
)
elif param_type == "baseurl":
return BaseUrlParameterModel(
type="baseurl",
name=input_source.parse_name(),
**_common_param_kwargs(input_source),
)
elif param_type == "genomebuild":
optional = input_source.parse_optional()
@@ -292,6 +325,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
name=input_source.parse_name(),
optional=optional,
multiple=multiple,
**_common_param_kwargs(input_source),
)
elif param_type == "directory_uri":
directory_uri_validators: List[TextCompatiableValidators] = _text_validators(input_source)
@@ -299,6 +333,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
type="directory",
name=input_source.parse_name(),
validators=directory_uri_validators,
**_common_param_kwargs(input_source),
)
else:
raise UnknownParameterTypeError(f"Unknown Galaxy parameter type {param_type}")
@@ -336,6 +371,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
name=input_source.parse_name(),
test_parameter=test_parameter,
whens=whens,
**_common_param_kwargs(input_source),
)
elif input_type == "repeat":
name = input_source.get("name")
@@ -353,6 +389,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
parameters=instance_tool_parameter_models,
min=min,
max=max,
**_common_param_kwargs(input_source),
)
elif input_type == "section":
name = input_source.get("name")
@@ -362,6 +399,7 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
type="section",
name=name,
parameters=instance_tool_parameter_models,
**_common_param_kwargs(input_source),
)
else:
raise Exception(
lib/galaxy/tool_util_models/parameters.py
+127 -5
View File
@@ -158,6 +158,10 @@ class ParamModel(Protocol):
# input value MUST be specified.
...
def field_kwargs(self) -> Dict[str, Any]:
"""Return kwargs for pydantic Field() including json_schema_extra metadata."""
...
def safe_field_name(name: str) -> str:
if name.startswith("_"):
@@ -165,6 +169,10 @@ def safe_field_name(name: str) -> str:
return name
def _label_value_dicts(options: List[Any]) -> List[Dict[str, Any]]:
return [{"label": o.label, "value": o.value, "selected": o.selected} for o in options]
def dynamic_model_information_from_py_type(
param_model: ParamModel, py_type: Type, requires_value: Optional[bool] = None, validators=None
):
@@ -177,9 +185,10 @@ def dynamic_model_information_from_py_type(
if not py_type_is_optional and not requires_value:
validators["not_null"] = field_validator(name)(Validators.validate_not_none)
field_kwargs = param_model.field_kwargs()
return DynamicModelInformation(
name,
(py_type, Field(initialize, alias=param_model.name if param_model.name != name else None)),
(py_type, Field(initialize, alias=param_model.name if param_model.name != name else None, **field_kwargs)),
validators,
)
@@ -198,6 +207,10 @@ class BaseToolParameterModelDefinition(ToolSourceBaseModel):
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
"""Return info needed to build Pydantic model at runtime for validation."""
def field_kwargs(self) -> Dict[str, Any]:
"""Return kwargs for pydantic Field() including json_schema_extra metadata."""
return {"json_schema_extra": {"gx_type": self.parameter_type}}
class BaseGalaxyToolParameterModelDefinition(BaseToolParameterModelDefinition):
hidden: bool = False
@@ -219,6 +232,20 @@ class BaseGalaxyToolParameterModelDefinition(BaseToolParameterModelDefinition):
is_dynamic: bool = False
optional: Annotated[bool, Field(description="If `false`, parameter must have a value.")] = False
def field_kwargs(self) -> Dict[str, Any]:
kwargs: Dict[str, Any] = {}
if self.label:
kwargs["title"] = self.label
description_parts = []
if self.help:
description_parts.append(self.help)
if self.argument:
description_parts.append(f"({self.argument})")
if description_parts:
kwargs["description"] = " ".join(description_parts)
kwargs["json_schema_extra"] = {"gx_type": self.parameter_type}
return kwargs
class LabelValue(BaseModel):
label: str
@@ -279,6 +306,14 @@ class TextParameterModel(BaseGalaxyToolParameterModelDefinition):
default_options: List[LabelValue] = []
validators: List[TextCompatiableValidators] = []
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
extra = kwargs["json_schema_extra"]
extra["gx_area"] = self.area
if self.default_options:
extra["gx_default_options"] = _label_value_dicts(self.default_options)
return kwargs
@property
def py_type(self) -> Type:
return optional_if_needed(StrictStr, self.optional)
@@ -318,6 +353,15 @@ class IntegerParameterModel(BaseGalaxyToolParameterModelDefinition):
max: Optional[int] = None
validators: List[NumberCompatiableValidators] = []
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
extra = kwargs["json_schema_extra"]
if self.min is not None:
extra["gx_min"] = self.min
if self.max is not None:
extra["gx_max"] = self.max
return kwargs
@property
def py_type(self) -> Type:
return optional_if_needed(StrictInt, self.optional)
@@ -350,6 +394,15 @@ class FloatParameterModel(BaseGalaxyToolParameterModelDefinition):
max: Optional[float] = None
validators: List[NumberCompatiableValidators] = []
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
extra = kwargs["json_schema_extra"]
if self.min is not None:
extra["gx_min"] = self.min
if self.max is not None:
extra["gx_max"] = self.max
return kwargs
@property
def py_type(self) -> Type:
return optional_if_needed(union_type([StrictInt, StrictFloat]), self.optional)
@@ -919,6 +972,17 @@ class DataParameterModel(BaseGalaxyToolParameterModelDefinition):
min: Optional[int] = None
max: Optional[int] = None
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
extra = kwargs["json_schema_extra"]
extra["gx_extensions"] = self.extensions
extra["gx_multiple"] = self.multiple
if self.min is not None:
extra["gx_min"] = self.min
if self.max is not None:
extra["gx_max"] = self.max
return kwargs
@property
def py_type(self) -> Type:
base_model: Type
@@ -1111,6 +1175,11 @@ class DataCollectionParameterModel(BaseGalaxyToolParameterModelDefinition):
extensions: List[str] = ["data"]
value: Optional[Dict[str, Any]]
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
kwargs["json_schema_extra"]["gx_extensions"] = self.extensions
return kwargs
@property
def py_type(self) -> Type:
return optional_if_needed(DataCollectionRequestOrCollectionUri, self.optional)
@@ -1294,9 +1363,10 @@ class ColorParameterModel(BaseGalaxyToolParameterModelDefinition):
validators = {"color_format": field_validator(self.name)(ColorParameterModel.validate_color_str_if_value)}
else:
validators = {"color_format": field_validator(self.name)(ColorParameterModel.validate_color_str)}
field_kwargs = self.field_kwargs()
return DynamicModelInformation(
self.name,
(py_type, initialize),
(py_type, Field(initialize, **field_kwargs)),
validators,
)
@@ -1392,6 +1462,14 @@ class SelectParameterModel(BaseGalaxyToolParameterModelDefinition):
multiple: bool = False
validators: List[SelectCompatiableValidators] = []
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
extra = kwargs["json_schema_extra"]
if self.options is not None and self.options:
extra["gx_options"] = _label_value_dicts(self.options)
extra["gx_multiple"] = self.multiple
return kwargs
@staticmethod
def split_str(cls, data: Any) -> Any:
if isinstance(data, str):
@@ -1498,6 +1576,11 @@ class GenomeBuildParameterModel(BaseGalaxyToolParameterModelDefinition):
type: Literal["genomebuild"]
multiple: bool
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
kwargs["json_schema_extra"]["gx_multiple"] = self.multiple
return kwargs
@property
def py_type(self) -> Type:
py_type: Type = StrictStr
@@ -1552,6 +1635,11 @@ class DrillDownParameterModel(BaseGalaxyToolParameterModelDefinition):
multiple: bool
hierarchy: DrillDownHierarchyT
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
kwargs["json_schema_extra"]["gx_multiple"] = self.multiple
return kwargs
@property
def py_type(self) -> Type:
if self.options is not None:
@@ -1645,6 +1733,11 @@ class DataColumnParameterModel(BaseGalaxyToolParameterModelDefinition):
multiple: bool
value: Optional[Union[int, List[int]]] = None
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
kwargs["json_schema_extra"]["gx_multiple"] = self.multiple
return kwargs
@staticmethod
def split_str(cls, data: Any) -> Any:
if isinstance(data, str):
@@ -1695,6 +1788,11 @@ class GroupTagParameterModel(BaseGalaxyToolParameterModelDefinition):
type: Literal["group_tag"]
multiple: bool
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
kwargs["json_schema_extra"]["gx_multiple"] = self.multiple
return kwargs
@property
def py_type(self) -> Type:
py_type: Type = StrictStr
@@ -1757,6 +1855,18 @@ class ConditionalParameterModel(BaseGalaxyToolParameterModelDefinition):
test_parameter: Union[BooleanParameterModel, SelectParameterModel]
whens: List[ConditionalWhen]
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
extra = kwargs["json_schema_extra"]
test_param = self.test_parameter
if isinstance(test_param, SelectParameterModel) and test_param.options:
extra["gx_options"] = _label_value_dicts(test_param.options)
if test_param.label:
extra["gx_test_label"] = test_param.label
if test_param.help:
extra["gx_test_help"] = test_param.help
return kwargs
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
is_boolean = isinstance(self.test_parameter, BooleanParameterModel)
test_param_name = self.test_parameter.name
@@ -1846,9 +1956,10 @@ class ConditionalParameterModel(BaseGalaxyToolParameterModelDefinition):
else:
initialize_cond = None
field_kwargs = self.field_kwargs()
return DynamicModelInformation(
self.name,
(py_type, initialize_cond),
(py_type, Field(initialize_cond, **field_kwargs)),
{},
)
@@ -1864,6 +1975,15 @@ class RepeatParameterModel(BaseGalaxyToolParameterModelDefinition):
min: Optional[int] = None
max: Optional[int] = None
def field_kwargs(self) -> Dict[str, Any]:
kwargs = super().field_kwargs()
extra = kwargs["json_schema_extra"]
if self.min is not None:
extra["gx_min"] = self.min
if self.max is not None:
extra["gx_max"] = self.max
return kwargs
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
# Maybe validators for min and max...
instance_class: Type[BaseModel] = create_field_model(
@@ -1887,9 +2007,10 @@ class RepeatParameterModel(BaseGalaxyToolParameterModelDefinition):
class RepeatType(RootModel):
root: List[instance_class] = Field(initialize_repeat, min_length=min_length, max_length=max_length) # type: ignore[valid-type]
field_kwargs = self.field_kwargs()
return DynamicModelInformation(
self.name,
(RepeatType, initialize_repeat),
(RepeatType, Field(initialize_repeat, **field_kwargs)),
{},
)
@@ -1921,9 +2042,10 @@ class SectionParameterModel(BaseGalaxyToolParameterModelDefinition):
initialize_section = ...
else:
initialize_section = None
field_kwargs = self.field_kwargs()
return DynamicModelInformation(
self.name,
(instance_class, initialize_section),
(instance_class, Field(initialize_section, **field_kwargs)),
{},
)