Skip to content

Commit

Permalink
Pydantic models for parameter validators.
Browse files Browse the repository at this point in the history
  • Loading branch information
@jmchilton
jmchilton committed Oct 8, 2024
1 parent bcf9fab commit 90ce472
Show file tree
Hide file tree
Showing 26 changed files with 1,466 additions and 301 deletions.
11 changes: 11 additions & 0 deletions lib/galaxy/tool_util/parameters/_types.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
"""

from typing import (
Any,
cast,
List,
Optional,
Expand All @@ -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,
)
Expand Down Expand Up @@ -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]
67 changes: 64 additions & 3 deletions lib/galaxy/tool_util/parameters/factory.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,19 @@
PagesSource,
ToolSource,
)
from galaxy.tool_util.parser.util import parse_profile_version
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,
text_input_is_optional,
)
from galaxy.util import string_as_bool
from .models import (
BaseUrlParameterModel,
Expand Down Expand Up @@ -42,10 +54,13 @@
HiddenParameterModel,
IntegerParameterModel,
LabelValue,
NumberCompatiableValidators,
RepeatParameterModel,
RulesParameterModel,
SectionParameterModel,
SelectCompatiableValidators,
SelectParameterModel,
TextCompatiableValidators,
TextParameterModel,
ToolParameterBundle,
ToolParameterBundleModel,
Expand Down Expand Up @@ -82,7 +97,23 @@ 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))
min_raw = input_source.get("min", None)
max_raw = input_source.get("max", None)
min_int = int(min_raw) if min_raw is not None else None
max_int = int(max_raw) if max_raw is not None else None
return IntegerParameterModel(
name=input_source.parse_name(),
optional=optional,
value=int_value,
min=min_int,
max=max_int,
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)
Expand All @@ -92,10 +123,22 @@ def _from_input_source_galaxy(input_source: InputSource, profile: float) -> Tool
value=value,
)
elif param_type == "text":
optional = input_source.parse_optional()
optional, optionality_inferred = text_input_is_optional(input_source)
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()
Expand All @@ -107,10 +150,22 @@ 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))
min_raw = input_source.get("min", None)
max_raw = input_source.get("max", None)
min_float = float(min_raw) if min_raw is not None else None
max_float = float(max_raw) if max_raw is not None else None
return FloatParameterModel(
name=input_source.parse_name(),
optional=optional,
value=float_value,
min=min_float,
max=max_float,
validators=float_validators,
)
elif param_type == "hidden":
optional = input_source.parse_optional()
Expand Down Expand Up @@ -158,11 +213,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)
Expand Down
96 changes: 83 additions & 13 deletions lib/galaxy/tool_util/parameters/models.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,9 @@
Mapping,
NamedTuple,
Optional,
Sequence,
Type,
TypeVar,
Union,
)

Expand All @@ -24,6 +26,7 @@
Field,
field_validator,
HttpUrl,
PlainValidator,
RootModel,
StrictBool,
StrictFloat,
Expand All @@ -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,
list_type,
optional,
Expand Down Expand Up @@ -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:
Expand All @@ -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:
Expand All @@ -223,6 +271,12 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
requires_value = True
elif _is_landing_request(state_representation):
requires_value = False
validators = self.validators[:]
if self.min is not None or self.max is not None:
validators.append(InRangeParameterValidatorModel(min=self.min, max=self.max, implicit=True))
pydantic_validators = static_tool_validators_to_pydantic(validators)
if pydantic_validators:
py_type = expand_annotation(py_type, pydantic_validators)
return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value)

@property
Expand All @@ -235,6 +289,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:
Expand All @@ -249,6 +304,12 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
requires_value = True
elif _is_landing_request(state_representation):
requires_value = False
validators = self.validators[:]
if self.min is not None or self.max is not None:
validators.append(InRangeParameterValidatorModel(min=self.min, max=self.max, implicit=True))
pydantic_validators = static_tool_validators_to_pydantic(validators)
if pydantic_validators:
py_type = expand_annotation(py_type, pydantic_validators)
return dynamic_model_information_from_py_type(self, py_type, requires_value=requires_value)

@property
Expand Down Expand Up @@ -659,10 +720,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:
Expand Down Expand Up @@ -699,28 +764,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):
Expand Down
3 changes: 2 additions & 1 deletion lib/galaxy/tool_util/parser/interface.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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.
Expand Down
Loading

0 comments on commit 90ce472

Please sign in to comment.