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

Harden collection runtime models with discriminated unions.

Browse Source 
Use Pydantic discriminated unions to enforce that collection_type
matches element structure:
- Add Literal types to simple models (list, paired, record, etc.)
- Add DataCollectionSampleSheetRuntime as separate model
- Split DataCollectionNestedRuntime into NestedList and NestedRecord
- Add collection_runtime_discriminator() for routing validation
- Create CollectionRuntimeDiscriminated tagged union
- Add structure mismatch tests (list with object elements, etc.)
- Add gx_data_collection_list_or_paired test tool

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
John Chilton
2026-02-12 15:58:17 -05:00
co-authored by Claude Opus 4.5
parent 0edbe4df31
commit 3f3a656535
6 changed files with 406 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/tool_util_models/parameters.py
+172 -44
View File
@@ -638,46 +638,134 @@ class DataCollectionPairedElements(StrictModel):
class DataCollectionPairedRuntime(DataCollectionInternalJsonBase):
"""Paired collection runtime representation."""
collection_type: Literal["paired"]
elements: DataCollectionPairedElements
class DataCollectionListRuntime(DataCollectionInternalJsonBase):
"""List collection runtime representation."""
collection_type: Literal["list"]
elements: List[DataCollectionElementInternalJson]
class DataCollectionSampleSheetRuntime(DataCollectionInternalJsonBase):
"""Sample sheet collection runtime representation."""
collection_type: Literal["sample_sheet"]
elements: List[DataCollectionElementInternalJson]
class DataCollectionRecordRuntime(DataCollectionInternalJsonBase):
"""Record collection runtime representation."""
elements: Dict[str, Union[DataCollectionElementInternalJson, "DataCollectionNestedRuntime"]]
collection_type: Literal["record"]
elements: Dict[str, Union[DataCollectionElementInternalJson, "DataCollectionNestedListRuntime", "DataCollectionNestedRecordRuntime"]]
class DataCollectionPairedOrUnpairedRuntime(DataCollectionInternalJsonBase):
"""Paired or Unpaired collection runtime representation."""
collection_type: Literal["paired_or_unpaired"]
elements: Dict[str, DataCollectionElementInternalJson]
class DataCollectionNestedRuntime(DataCollectionInternalJsonBase):
"""Nested collection runtime representation (list:paired, etc.)."""
elements: Union[
List[Union[
"DataCollectionListRuntime",
"DataCollectionPairedRuntime",
"DataCollectionRecordRuntime",
"DataCollectionPairedOrUnpairedRuntime",
"DataCollectionNestedRuntime",
]],
Dict[str, Union[
DataCollectionElementInternalJson,
"DataCollectionListRuntime",
"DataCollectionPairedRuntime",
"DataCollectionRecordRuntime",
"DataCollectionPairedOrUnpairedRuntime",
"DataCollectionNestedRuntime",
]]
]
class DataCollectionNestedListRuntime(DataCollectionInternalJsonBase):
"""Nested collection with list-like outer structure (list:*, sample_sheet:*)."""
collection_type: str
@field_validator('collection_type')
@classmethod
def must_be_nested_list_like(cls, v: str) -> str:
if ':' not in v:
raise ValueError(f'Nested collection_type must contain ":", got "{v}"')
first_segment = v.split(':')[0]
if first_segment not in ('list', 'sample_sheet'):
raise ValueError(f'Outer type must be list-like (list, sample_sheet), got "{first_segment}"')
return v
elements: List[Union[
"DataCollectionListRuntime",
"DataCollectionSampleSheetRuntime",
"DataCollectionPairedRuntime",
"DataCollectionRecordRuntime",
"DataCollectionPairedOrUnpairedRuntime",
"DataCollectionNestedListRuntime",
"DataCollectionNestedRecordRuntime",
]]
DataCollectionNestedRuntime.model_rebuild()
class DataCollectionNestedRecordRuntime(DataCollectionInternalJsonBase):
"""Nested collection with record-like outer structure (paired:*, record:*)."""
collection_type: str
@field_validator('collection_type')
@classmethod
def must_be_nested_record_like(cls, v: str) -> str:
if ':' not in v:
raise ValueError(f'Nested collection_type must contain ":", got "{v}"')
first_segment = v.split(':')[0]
if first_segment in ('list', 'sample_sheet'):
raise ValueError(f'Outer type must be record-like, got list-like "{first_segment}"')
return v
elements: Dict[str, Union[
DataCollectionElementInternalJson,
"DataCollectionListRuntime",
"DataCollectionSampleSheetRuntime",
"DataCollectionPairedRuntime",
"DataCollectionRecordRuntime",
"DataCollectionPairedOrUnpairedRuntime",
"DataCollectionNestedListRuntime",
"DataCollectionNestedRecordRuntime",
]]
DataCollectionNestedListRuntime.model_rebuild()
DataCollectionNestedRecordRuntime.model_rebuild()
def collection_runtime_discriminator(v: Any) -> str:
"""Discriminator function for collection runtime unions.
Routes validation to the correct model based on collection_type pattern.
"""
if isinstance(v, dict):
ct = v.get('collection_type', '')
else:
ct = getattr(v, 'collection_type', '')
# Simple types - exact match
if ct == 'list':
return 'list'
elif ct == 'paired':
return 'paired'
elif ct == 'record':
return 'record'
elif ct == 'paired_or_unpaired':
return 'paired_or_unpaired'
elif ct == 'sample_sheet':
return 'sample_sheet'
elif ':' in ct:
# Nested types - route by outer structure
first_segment = ct.split(':')[0]
if first_segment in ('list', 'sample_sheet'):
return 'nested_list'
else:
return 'nested_record'
else:
# Unknown type - default to list (matches historical behavior)
return 'list'
CollectionRuntimeDiscriminated: Type = Annotated[
Union[
Annotated[DataCollectionListRuntime, Tag('list')],
Annotated[DataCollectionSampleSheetRuntime, Tag('sample_sheet')],
Annotated[DataCollectionPairedRuntime, Tag('paired')],
Annotated[DataCollectionRecordRuntime, Tag('record')],
Annotated[DataCollectionPairedOrUnpairedRuntime, Tag('paired_or_unpaired')],
Annotated[DataCollectionNestedListRuntime, Tag('nested_list')],
Annotated[DataCollectionNestedRecordRuntime, Tag('nested_record')],
],
Discriminator(collection_runtime_discriminator)
]
DataRequestInternal: Type = cast(
@@ -945,32 +1033,72 @@ class DataCollectionParameterModel(BaseGalaxyToolParameterModelDefinition):
def py_type_internal_dereferenced(self) -> Type:
return optional_if_needed(DataCollectionRequestInternal, self.optional)
def _runtime_model_for_collection_type(self, ct: str) -> tuple:
"""Map a single collection type to its runtime model and tag.
Returns tuple of (model, tag) for use in discriminated unions.
"""
if ct == "list":
return (DataCollectionListRuntime, 'list')
elif ct == "paired":
return (DataCollectionPairedRuntime, 'paired')
elif ct == "record":
return (DataCollectionRecordRuntime, 'record')
elif ct == "paired_or_unpaired":
return (DataCollectionPairedOrUnpairedRuntime, 'paired_or_unpaired')
elif ct == "sample_sheet":
return (DataCollectionSampleSheetRuntime, 'sample_sheet')
elif ":" in ct:
# Nested types like "list:paired", "sample_sheet:paired", etc.
first_segment = ct.split(":")[0]
if first_segment in ("list", "sample_sheet"):
return (DataCollectionNestedListRuntime, 'nested_list')
else:
return (DataCollectionNestedRecordRuntime, 'nested_record')
else:
# Unknown single type - return None to signal fallback needed
return (None, None)
@property
def py_type_internal_json(self) -> Type:
# Return normalized collection runtime models with metadata
if self.collection_type == "list":
return optional_if_needed(DataCollectionListRuntime, self.optional)
elif self.collection_type == "paired":
return optional_if_needed(DataCollectionPairedRuntime, self.optional)
elif self.collection_type == "record":
return optional_if_needed(DataCollectionRecordRuntime, self.optional)
elif self.collection_type == "paired_or_unpaired":
return optional_if_needed(DataCollectionPairedOrUnpairedRuntime, self.optional)
elif self.collection_type == "sample_sheet":
return optional_if_needed(DataCollectionListRuntime, self.optional)
elif self.collection_type:
# Nested types like "list:paired", "sample_sheet:paired", etc.
return optional_if_needed(DataCollectionNestedRuntime, self.optional)
else:
# Unknown collection_type - union of all runtime types
base_type = union_type([
DataCollectionListRuntime,
DataCollectionPairedRuntime,
DataCollectionRecordRuntime,
DataCollectionPairedOrUnpairedRuntime,
DataCollectionNestedRuntime,
])
return optional_if_needed(base_type, self.optional)
if not self.collection_type:
# Unknown collection_type - use full discriminated union
return optional_if_needed(CollectionRuntimeDiscriminated, self.optional)
# Handle comma-separated collection types (e.g., "list,paired")
if "," in self.collection_type:
types = [t.strip() for t in self.collection_type.split(",")]
tagged_types = []
tags_seen: set = set()
for t in types:
model, tag_str = self._runtime_model_for_collection_type(t)
if model and tag_str not in tags_seen:
tags_seen.add(tag_str)
tagged_types.append(Annotated[model, Tag(tag_str)])
if tagged_types:
if len(tagged_types) == 1:
# Single type - no union needed
base_type = tagged_types[0].__origin__ # Get unwrapped type
else:
# Multiple types - build discriminated union
subset_union = Annotated[
Union[tuple(tagged_types)],
Discriminator(collection_runtime_discriminator)
]
base_type = subset_union
return optional_if_needed(base_type, self.optional)
# Fall through to full union if no models matched
# Single collection type
model, _tag = self._runtime_model_for_collection_type(self.collection_type)
if model:
return optional_if_needed(model, self.optional)
# Fallback for unrecognized types - use full discriminated union
return optional_if_needed(CollectionRuntimeDiscriminated, self.optional)
def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
if state_representation in ["request", "relaxed_request"]:
lib/galaxy/tools/runtime.py
+27 -13
View File
@@ -21,10 +21,12 @@ from galaxy.tool_util_models.parameters import (
DataRequestInternalDereferencedT,
DataRequestInternalHda,
DataCollectionListRuntime,
DataCollectionNestedRuntime,
DataCollectionNestedListRuntime,
DataCollectionNestedRecordRuntime,
DataCollectionPairedRuntime,
DataCollectionPairedOrUnpairedRuntime,
DataCollectionRecordRuntime,
DataCollectionSampleSheetRuntime,
)
if TYPE_CHECKING:
@@ -35,10 +37,12 @@ if TYPE_CHECKING:
# Type aliases for callbacks
CollectionRuntimeT = Union[
DataCollectionListRuntime,
DataCollectionSampleSheetRuntime,
DataCollectionPairedRuntime,
DataCollectionRecordRuntime,
DataCollectionPairedOrUnpairedRuntime,
DataCollectionNestedRuntime,
DataCollectionNestedListRuntime,
DataCollectionNestedRecordRuntime,
]
DatasetToRuntimeJson = Callable[[DataRequestInternalDereferencedT], DataInternalJson]
CollectionToRuntimeJson = Callable[[DataCollectionRequestInternal, Optional[str]], CollectionRuntimeT]
@@ -182,24 +186,34 @@ def collection_to_runtime(
def _validate_collection_runtime_dict(raw: Dict[str, Any]) -> CollectionRuntimeT:
"""Parse raw dict into appropriate typed collection runtime model."""
"""Parse raw dict into appropriate typed collection runtime model.
Routes to the correct model based on collection_type, enforcing that
the element structure matches what the collection_type claims.
"""
ct = raw.get("collection_type", "")
if ct == "paired":
# Simple types - exact match with Literal enforcement
if ct == "list":
return DataCollectionListRuntime.model_validate(raw)
elif ct == "sample_sheet":
return DataCollectionSampleSheetRuntime.model_validate(raw)
elif ct == "paired":
return DataCollectionPairedRuntime.model_validate(raw)
elif ct == "record":
return DataCollectionRecordRuntime.model_validate(raw)
elif ct == "paired_or_unpaired":
return DataCollectionPairedOrUnpairedRuntime.model_validate(raw)
elif not ct or ":" in ct:
# Nested types (list:paired, etc.) or unknown
return DataCollectionNestedRuntime.model_validate(raw)
else:
# Simple list-like types (list, sample_sheet, record, paired_or_unpaired, etc.)
# For list/sample_sheet: elements is List[DataCollectionElementInternalJson]
if is_list_like(ct):
return DataCollectionListRuntime.model_validate(raw)
elif ":" in ct:
# Nested types - route by outer structure
first_segment = ct.split(":")[0]
if first_segment in ("list", "sample_sheet"):
return DataCollectionNestedListRuntime.model_validate(raw)
else:
return DataCollectionNestedRuntime.model_validate(raw)
return DataCollectionNestedRecordRuntime.model_validate(raw)
else:
# Unknown type - try list (historical behavior)
return DataCollectionListRuntime.model_validate(raw)
def _build_collection_runtime_dict(
test/functional/tools/parameters/gx_data_collection_list_or_paired.yml
+50
View File
@@ -0,0 +1,50 @@
class: GalaxyTool
id: gx_data_collection_list_or_paired
name: gx_data_collection_list_or_paired
version: "1.0.0"
container: busybox
shell_command: |
echo "Collection type: $(inputs.parameter.collection_type)" > output.txt
echo "Name: $(inputs.parameter.name)" >> output.txt
inputs:
- name: parameter
type: data_collection
collection_type: list,paired
label: Input list or paired collection
outputs:
- name: output
type: data
format: txt
from_work_dir: output.txt
tests:
- inputs:
parameter:
class: Collection
collection_type: list
name: test_list
elements:
- identifier: f1
class: File
path: simple_line.txt
outputs:
output:
asserts:
- has_line:
line: "Collection type: list"
- inputs:
parameter:
class: Collection
collection_type: paired
name: test_paired
elements:
- identifier: forward
class: File
path: simple_line.txt
- identifier: reverse
class: File
path: simple_line_alternative.txt
outputs:
output:
asserts:
- has_line:
line: "Collection type: paired"
test/functional/tools/sample_tool_conf.xml
+1
View File
@@ -296,6 +296,7 @@
<tool file="parameters/gx_data_user.yml" />
<tool file="parameters/gx_data_multiple_user.yml" />
<tool file="parameters/gx_select_multiple_one_default_user.yml" />
<tool file="parameters/gx_data_collection_list_or_paired.yml" />
<section id="filter" name="For Tours">
<tool file="for_tours/filtering.xml" />
test/unit/tool_util/parameter_specification.yml
+151
View File
@@ -2460,6 +2460,23 @@ gx_data_collection_list:
nameext: ".txt"
format: "txt"
size: 100
# Structure mismatch: collection_type=list but elements is object
- parameter:
class: Collection
name: list_wrong_structure
collection_type: list
tags: []
elements:
forward:
class: File
element_identifier: forward
basename: "f.txt"
location: "step_input://0"
path: "/tmp/f.txt"
nameroot: "f"
nameext: ".txt"
format: "txt"
size: 100
gx_data_collection_paired:
request_valid:
@@ -2542,6 +2559,140 @@ gx_data_collection_paired:
nameext: ".txt"
format: "txt"
size: 100
# Structure mismatch: collection_type=paired but elements is array
- parameter:
class: Collection
name: paired_wrong_structure
collection_type: paired
tags: []
elements:
- class: File
element_identifier: f1
basename: "f1.txt"
location: "step_input://0"
path: "/tmp/f1.txt"
nameroot: "f1"
nameext: ".txt"
format: "txt"
size: 100
gx_data_collection_list_or_paired:
request_valid:
- parameter: {src: hdca, id: abcdabcd}
request_invalid:
- parameter: {src: hdca, id: 7}
- parameter: null
- {}
request_internal_valid:
- parameter: {src: hdca, id: 5}
request_internal_invalid:
- parameter: {src: hdca, id: abcdabcd}
- parameter: null
- {}
job_internal_valid:
- parameter: {src: hdca, id: 5}
job_internal_invalid:
- parameter: {src: hdca, id: abcdabcd}
- {}
job_runtime_valid:
# List collection is valid
- parameter:
class: Collection
name: test_list
collection_type: list
tags: []
elements:
- class: File
element_identifier: f1
basename: "f1.txt"
location: "step_input://0"
path: "/tmp/f1.txt"
nameroot: "f1"
nameext: ".txt"
format: "txt"
size: 100
# Paired collection is also valid
- parameter:
class: Collection
name: test_paired
collection_type: paired
tags: []
elements:
forward:
class: File
element_identifier: forward
basename: "f.txt"
location: "step_input://0"
path: "/tmp/f.txt"
nameroot: "f"
nameext: ".txt"
format: "txt"
size: 100
reverse:
class: File
element_identifier: reverse
basename: "r.txt"
location: "step_input://1"
path: "/tmp/r.txt"
nameroot: "r"
nameext: ".txt"
format: "txt"
size: 100
job_runtime_invalid:
- parameter: {src: hdca, id: 5}
- {}
# Record collection type should be rejected (not list or paired)
- parameter:
class: Collection
name: wrong_type
collection_type: record
tags: []
elements:
field1:
class: File
element_identifier: field1
basename: "f1.txt"
location: "step_input://0"
path: "/tmp/f1.txt"
nameroot: "f1"
nameext: ".txt"
format: "txt"
size: 100
# Structure mismatch: collection_type=list but elements is object (paired structure)
# With discriminated unions, this should be properly rejected
- parameter:
class: Collection
name: list_wrong_structure
collection_type: list
tags: []
elements:
forward:
class: File
element_identifier: forward
basename: "f.txt"
location: "step_input://0"
path: "/tmp/f.txt"
nameroot: "f"
nameext: ".txt"
format: "txt"
size: 100
# Structure mismatch: collection_type=paired but elements is array (list structure)
# With discriminated unions, this should be properly rejected
- parameter:
class: Collection
name: paired_wrong_structure
collection_type: paired
tags: []
elements:
- class: File
element_identifier: f1
basename: "f1.txt"
location: "step_input://0"
path: "/tmp/f1.txt"
nameroot: "f1"
nameext: ".txt"
format: "txt"
size: 100
cwl_int:
request_valid:
test/unit/tool_util/test_runtime.py
+5 -4
View File
@@ -4,9 +4,10 @@ from galaxy.model import DatasetCollection, DatasetCollectionElement, HistoryDat
from galaxy.tool_util_models.parameters import (
DataCollectionPairedRuntime,
DataCollectionListRuntime,
DataCollectionNestedRuntime,
DataCollectionNestedListRuntime,
DataCollectionRecordRuntime,
DataCollectionPairedOrUnpairedRuntime,
DataCollectionSampleSheetRuntime,
DataInternalJson,
)
from galaxy.tools.runtime import collection_to_runtime
@@ -120,7 +121,7 @@ def test_collection_to_runtime_nested(mock_adapt_dataset):
compute_environment=None
)
assert isinstance(runtime, DataCollectionNestedRuntime)
assert isinstance(runtime, DataCollectionNestedListRuntime)
assert runtime.collection_type == "list:paired"
assert len(runtime.elements) == 1
@@ -179,8 +180,8 @@ def test_collection_to_runtime_sample_sheet(mock_adapt_dataset):
compute_environment=None
)
# Should use DataCollectionListRuntime but preserve metadata
assert isinstance(runtime, DataCollectionListRuntime)
# Should use DataCollectionSampleSheetRuntime to preserve metadata
assert isinstance(runtime, DataCollectionSampleSheetRuntime)
assert runtime.collection_type == "sample_sheet"
assert runtime.column_definitions is not None
assert runtime.elements[0].columns == ["treatment"]