Pydantic models for parameter validators.

lib/galaxy/tool_util/parameters/_types.py

@@ -6,6 +6,7 @@
 """
 
 from typing import (
+    Any,
     cast,
     List,
     Optional,
@@ -15,6 +16,7 @@
 
 # https://stackoverflow.com/questions/56832881/check-if-a-field-is-typing-optional
 from typing_extensions import (
+    Annotated,
     get_args,
     get_origin,
 )
@@ -46,3 +48,12 @@ def cast_as_type(arg) -> Type:
 
 def is_optional(field) -> bool:
     return get_origin(field) is Union and type(None) in get_args(field)
+
+
+def expand_annotation(field: Type, new_annotations: List[Any]) -> Type:
+    is_annotation = get_origin(field) is Annotated
+    if is_annotation:
+        args = get_args(field)  # noqa: F841
+        return Annotated[tuple([args[0], *args[1:], *new_annotations])]  # type: ignore[return-value]
+    else:
+        return Annotated[tuple([field, *new_annotations])]  # type: ignore[return-value]

lib/galaxy/tool_util/parameters/factory.py

@@ -14,6 +14,15 @@
     PagesSource,
     ToolSource,
 )
+from galaxy.tool_util.parser.parameter_validators import (
+    EmptyFieldParameterValidatorModel,
+    ExpressionParameterValidatorModel,
+    InRangeParameterValidatorModel,
+    LengthParameterValidatorModel,
+    NoOptionsParameterValidatorModel,
+    RegexParameterValidatorModel,
+    static_validators,
+)
 from galaxy.tool_util.parser.util import parse_profile_version
 from galaxy.util import string_as_bool
 from .models import (
@@ -42,10 +51,13 @@
     HiddenParameterModel,
     IntegerParameterModel,
     LabelValue,
+    NumberCompatiableValidators,
     RepeatParameterModel,
     RulesParameterModel,
     SectionParameterModel,
+    SelectCompatiableValidators,
     SelectParameterModel,
+    TextCompatiableValidators,
     TextParameterModel,
     ToolParameterBundle,
     ToolParameterBundleModel,
@@ -82,7 +94,14 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
                 int_value = None
             else:
                 raise ParameterDefinitionError()
-            return IntegerParameterModel(name=input_source.parse_name(), optional=optional, value=int_value)
+            static_validator_models = static_validators(input_source.parse_validators())
+            int_validators: List[NumberCompatiableValidators] = []
+            for static_validator in static_validator_models:
+                if static_validator.type == "in_range":
+                    int_validators.append(cast(InRangeParameterValidatorModel, static_validator))
+            return IntegerParameterModel(
+                name=input_source.parse_name(), optional=optional, value=int_value, validators=int_validators
+            )
         elif param_type == "boolean":
             nullable = input_source.parse_optional()
             value = input_source.get_bool_or_none("checked", None if nullable else False)
@@ -93,9 +112,21 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
             )
         elif param_type == "text":
             optional = input_source.parse_optional()
+            static_validator_models = static_validators(input_source.parse_validators())
+            text_validators: List[TextCompatiableValidators] = []
+            for static_validator in static_validator_models:
+                if static_validator.type == "length":
+                    text_validators.append(cast(LengthParameterValidatorModel, static_validator))
+                elif static_validator.type == "regex":
+                    text_validators.append(cast(RegexParameterValidatorModel, static_validator))
+                elif static_validator.type == "expression":
+                    text_validators.append(cast(ExpressionParameterValidatorModel, static_validator))
+                elif static_validator.type == "empty_field":
+                    text_validators.append(cast(EmptyFieldParameterValidatorModel, static_validator))
             return TextParameterModel(
                 name=input_source.parse_name(),
                 optional=optional,
+                validators=text_validators,
             )
         elif param_type == "float":
             optional = input_source.parse_optional()
@@ -107,10 +138,16 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
                 float_value = None
             else:
                 raise ParameterDefinitionError()
+            static_validator_models = static_validators(input_source.parse_validators())
+            float_validators: List[NumberCompatiableValidators] = []
+            for static_validator in static_validator_models:
+                if static_validator.type == "in_range":
+                    float_validators.append(cast(InRangeParameterValidatorModel, static_validator))
             return FloatParameterModel(
                 name=input_source.parse_name(),
                 optional=optional,
                 value=float_value,
+                validators=float_validators,
             )
         elif param_type == "hidden":
             optional = input_source.parse_optional()
@@ -158,11 +195,17 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
                 options = []
                 for option_label, option_value, selected in input_source.parse_static_options():
                     options.append(LabelValue(label=option_label, value=option_value, selected=selected))
+            static_validator_models = static_validators(input_source.parse_validators())
+            select_validators: List[SelectCompatiableValidators] = []
+            for static_validator in static_validator_models:
+                if static_validator.type == "no_options":
+                    select_validators.append(cast(NoOptionsParameterValidatorModel, static_validator))
             return SelectParameterModel(
                 name=input_source.parse_name(),
                 optional=optional,
                 options=options,
                 multiple=multiple,
+                validators=select_validators,
             )
         elif param_type == "drill_down":
             multiple = input_source.get_bool("multiple", False)

lib/galaxy/tool_util/parameters/models.py

@@ -11,7 +11,9 @@
     Mapping,
     NamedTuple,
     Optional,
+    Sequence,
     Type,
+    TypeVar,
     Union,
 )
 
@@ -24,6 +26,7 @@
     Field,
     field_validator,
     HttpUrl,
+    PlainValidator,
     RootModel,
     StrictBool,
     StrictFloat,
@@ -44,8 +47,18 @@
     JsonTestCollectionDefDict,
     JsonTestDatasetDefDict,
 )
+from galaxy.tool_util.parser.parameter_validators import (
+    EmptyFieldParameterValidatorModel,
+    ExpressionParameterValidatorModel,
+    InRangeParameterValidatorModel,
+    LengthParameterValidatorModel,
+    NoOptionsParameterValidatorModel,
+    RegexParameterValidatorModel,
+    StaticValidatorModel,
+)
 from ._types import (
     cast_as_type,
+    expand_annotation,
     is_optional as is_python_type_optional,
     list_type,
     optional,
@@ -179,11 +192,39 @@ class LabelValue(BaseModel):
     selected: bool
 
 
+TextCompatiableValidators = Union[
+    LengthParameterValidatorModel,
+    RegexParameterValidatorModel,
+    ExpressionParameterValidatorModel,
+    EmptyFieldParameterValidatorModel,
+]
+
+
+def pydantic_validator_for(validator_model: StaticValidatorModel):
+
+    def validator(v: Any) -> Any:
+        validator_model.statically_validate(v)
+        return v
+
+    return PlainValidator(validator)
+
+
+VT = TypeVar("VT", bound=StaticValidatorModel)
+
+
+def static_tool_validators_to_pydantic(static_tool_param_validators: Sequence[VT]) -> List[VT]:
+    pydantic_validators = []
+    for static_validator in static_tool_param_validators:
+        pydantic_validators.append(pydantic_validator_for(static_validator))
+    return pydantic_validators
+
+
 class TextParameterModel(BaseGalaxyToolParameterModelDefinition):
     parameter_type: Literal["gx_text"] = "gx_text"
     area: bool = False
     default_value: Optional[str] = Field(default=None, alias="value")
     default_options: List[LabelValue] = []
+    validators: List[TextCompatiableValidators] = []
 
     @property
     def py_type(self) -> Type:
@@ -196,19 +237,26 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
         requires_value = self.request_requires_value
         if state_representation == "job_internal":
             requires_value = True
+        validators = static_tool_validators_to_pydantic(self.validators)
+        if validators:
+            py_type = expand_annotation(py_type, validators)
         return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value)
 
     @property
     def request_requires_value(self) -> bool:
         return False
 
 
+NumberCompatiableValidators = Union[InRangeParameterValidatorModel,]
+
+
 class IntegerParameterModel(BaseGalaxyToolParameterModelDefinition):
     parameter_type: Literal["gx_integer"] = "gx_integer"
     optional: bool
     value: Optional[int] = None
     min: Optional[int] = None
     max: Optional[int] = None
+    validators: List[NumberCompatiableValidators] = []
 
     @property
     def py_type(self) -> Type:
@@ -223,6 +271,9 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
             requires_value = True
         elif _is_landing_request(state_representation):
             requires_value = False
+        validators = static_tool_validators_to_pydantic(self.validators)
+        if validators:
+            py_type = expand_annotation(py_type, validators)
         return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value)
 
     @property
@@ -235,6 +286,7 @@ class FloatParameterModel(BaseGalaxyToolParameterModelDefinition):
     value: Optional[float] = None
     min: Optional[float] = None
     max: Optional[float] = None
+    validators: List[NumberCompatiableValidators] = []
 
     @property
     def py_type(self) -> Type:
@@ -244,6 +296,9 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
         py_type = self.py_type
         if state_representation == "workflow_step_linked":
             py_type = allow_connected_value(py_type)
+        validators = static_tool_validators_to_pydantic(self.validators)
+        if validators:
+            py_type = expand_annotation(py_type, validators)
         return dynamic_model_information_from_py_type(self, py_type)
 
     @property
@@ -654,10 +709,14 @@ def request_requires_value(self) -> bool:
         return True
 
 
+SelectCompatiableValidators = Union[NoOptionsParameterValidatorModel,]
+
+
 class SelectParameterModel(BaseGalaxyToolParameterModelDefinition):
     parameter_type: Literal["gx_select"] = "gx_select"
     options: Optional[List[LabelValue]] = None
     multiple: bool
+    validators: List[SelectCompatiableValidators]
 
     @staticmethod
     def split_str(cls, data: Any) -> Any:
@@ -694,28 +753,33 @@ def py_type_workflow_step(self) -> Type:
         return optional(self.py_type_if_required())
 
     def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
+        validators = {}
+        requires_value = self.request_requires_value
+        py_type = None
         if state_representation == "workflow_step":
-            return dynamic_model_information_from_py_type(self, self.py_type_workflow_step, requires_value=False)
+            py_type = self.py_type_workflow_step
         elif state_representation == "workflow_step_linked":
             py_type = self.py_type_if_required(allow_connections=True)
-            return dynamic_model_information_from_py_type(
-                self, optional_if_needed(py_type, self.optional or self.multiple)
-            )
+            py_type = optional_if_needed(py_type, self.optional or self.multiple)
         elif state_representation == "test_case_xml":
             # in a YAML test case representation this can be string, in XML we are still expecting a comma separated string
             py_type = self.py_type_if_required(allow_connections=False)
             if self.multiple:
                 validators = {"from_string": field_validator(self.name, mode="before")(SelectParameterModel.split_str)}
-            else:
-                validators = {}
-            return dynamic_model_information_from_py_type(
-                self, optional_if_needed(py_type, self.optional), validators=validators
-            )
+            py_type = optional_if_needed(py_type, self.optional)
+        elif state_representation == "job_internal":
+            requires_value = True
+            py_type = self.py_type
         else:
-            requires_value = self.request_requires_value
-            if state_representation == "job_internal":
-                requires_value = True
-            return dynamic_model_information_from_py_type(self, self.py_type, requires_value=requires_value)
+            py_type = self.py_type
+
+        validator_models = static_tool_validators_to_pydantic(self.validators)
+        if validator_models:
+            py_type = expand_annotation(py_type, validator_models)
+
+        return dynamic_model_information_from_py_type(
+            self, py_type, validators=validators, requires_value=requires_value
+        )
 
     @property
     def has_selected_static_option(self):

lib/galaxy/tool_util/parser/interface.py

@@ -28,6 +28,7 @@
 
 from galaxy.util import Element
 from galaxy.util.path import safe_walk
+from .parameter_validators import AnyValidatorModel
 from .util import _parse_name
 
 if TYPE_CHECKING:
@@ -502,7 +503,7 @@ def parse_sanitizer_elem(self):
         """
         return None
 
-    def parse_validator_elems(self):
+    def parse_validators(self) -> List[AnyValidatorModel]:
         """Return an XML description of sanitizers. This is a stop gap
         until we can rework galaxy.tools.parameters.validation to not
         explicitly depend on XML.