Spaces:
Running
Running
| from __future__ import annotations | |
| import copy | |
| import re | |
| from collections.abc import Mapping | |
| from pathlib import Path | |
| from typing import Any | |
| from pydantic import ValidationError | |
| from app.core.exceptions import TemplateValidationError | |
| from app.templates.models import PreparedTemplate, ResolvedPipelineStep | |
| from app.templates.schema import ( | |
| OutputDefinition, | |
| ParameterDefinition, | |
| ParameterType, | |
| TemplateDefinition, | |
| ) | |
| VARIABLE_PATTERN = re.compile(r"{{\s*([a-z][a-z0-9_]*)\s*}}") | |
| EXACT_VARIABLE_PATTERN = re.compile(r"^\s*{{\s*([a-z][a-z0-9_]*)\s*}}\s*$") | |
| INPUT_SELECTOR_PATTERN = re.compile( | |
| r"^(?:current|original|originals|original:[0-9]+|artifact:[a-z][a-z0-9_]*)$" | |
| ) | |
| class TemplateValidator: | |
| """Validates template definitions, parameters, variables, and operation names.""" | |
| def __init__(self, supported_operations: set[str]) -> None: | |
| self.supported_operations = frozenset(supported_operations) | |
| def validate_definition(self, template: TemplateDefinition) -> None: | |
| """Fail if a parsed YAML definition is unsafe or internally inconsistent.""" | |
| for name, definition in template.parameters.items(): | |
| if definition.has_default: | |
| self._validate_value(name, definition.default, definition) | |
| known_parameters = set(template.parameters) | |
| saved_artifacts: set[str] = set() | |
| for index, step in enumerate(template.pipeline): | |
| if step.operation not in self.supported_operations: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' uses unsupported operation '{step.operation}'", | |
| details={"step": index, "operation": step.operation}, | |
| ) | |
| referenced = self._variables(step.when) | self._variables(step.operation_parameters()) | |
| self._validate_variables(referenced, known_parameters, template.id, index) | |
| for parameter in referenced: | |
| definition = template.parameters.get(parameter) | |
| if definition and not definition.required and not definition.has_default: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' references optional parameter '{parameter}' without a default", | |
| details={"step": index}, | |
| ) | |
| for selector in step.inputs or []: | |
| if not INPUT_SELECTOR_PATTERN.fullmatch(selector): | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' has invalid input selector '{selector}'", | |
| details={"step": index}, | |
| ) | |
| if selector.startswith("artifact:"): | |
| artifact = selector.split(":", 1)[1] | |
| if artifact not in saved_artifacts: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' references artifact '{artifact}' before it is saved", | |
| details={"step": index}, | |
| ) | |
| if step.save_as: | |
| if step.save_as in saved_artifacts: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' saves duplicate artifact '{step.save_as}'" | |
| ) | |
| saved_artifacts.add(step.save_as) | |
| output_variables = self._variables(template.output.model_dump()) | |
| self._validate_variables(output_variables, known_parameters, template.id, -1) | |
| for parameter in output_variables: | |
| definition = template.parameters.get(parameter) | |
| if definition and not definition.required and not definition.has_default: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' output references optional parameter '{parameter}' without a default" | |
| ) | |
| if "{{" not in template.output.format: | |
| output_format = template.output.format.lower().lstrip(".") | |
| if output_format not in template.supported_outputs: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' output format is not declared in supported_outputs" | |
| ) | |
| def prepare( | |
| self, template: TemplateDefinition, supplied: Mapping[str, Any] | None | |
| ) -> PreparedTemplate: | |
| """Resolve validated runtime parameters and render a safe pipeline.""" | |
| parameters = self.resolve_parameters(template, supplied) | |
| steps: list[ResolvedPipelineStep] = [] | |
| for step in template.pipeline: | |
| rendered_when = self._substitute(step.when, parameters) | |
| if not isinstance(rendered_when, bool): | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' step condition must resolve to a boolean" | |
| ) | |
| rendered_parameters = self._substitute(step.operation_parameters(), parameters) | |
| steps.append( | |
| ResolvedPipelineStep( | |
| operation=step.operation, | |
| parameters=rendered_parameters, | |
| enabled=rendered_when, | |
| inputs=step.inputs, | |
| save_as=step.save_as, | |
| ) | |
| ) | |
| output_data = self._substitute(template.output.model_dump(), parameters) | |
| try: | |
| output = OutputDefinition.model_validate(output_data) | |
| except ValidationError as exc: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' output did not resolve to a valid contract" | |
| ) from exc | |
| output_format = output.format.strip().lower().lstrip(".") | |
| if not re.fullmatch(r"[a-z0-9][a-z0-9_+-]*", output_format): | |
| raise TemplateValidationError(f"Template '{template.id}' output format is invalid") | |
| if output_format not in template.supported_outputs: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' output format is not declared in supported_outputs" | |
| ) | |
| if output.filename: | |
| filename = output.filename | |
| if Path(filename).name != filename or len(filename) > 255: | |
| raise TemplateValidationError( | |
| f"Template '{template.id}' output filename must be a safe basename" | |
| ) | |
| return PreparedTemplate( | |
| definition=template, | |
| parameters=parameters, | |
| pipeline=tuple(steps), | |
| output=output, | |
| ) | |
| def resolve_parameters( | |
| self, template: TemplateDefinition, supplied: Mapping[str, Any] | None | |
| ) -> dict[str, Any]: | |
| """Apply defaults and enforce declared parameter names and strict types.""" | |
| provided = dict(supplied or {}) | |
| unknown = sorted(set(provided) - set(template.parameters)) | |
| if unknown: | |
| raise TemplateValidationError( | |
| "Unknown template parameter(s)", details={"parameters": unknown} | |
| ) | |
| resolved: dict[str, Any] = {} | |
| for name, definition in template.parameters.items(): | |
| if name in provided: | |
| value = provided[name] | |
| elif definition.has_default: | |
| value = copy.deepcopy(definition.default) | |
| elif definition.required: | |
| raise TemplateValidationError(f"Required template parameter '{name}' is missing") | |
| else: | |
| continue | |
| self._validate_value(name, value, definition) | |
| resolved[name] = value | |
| return resolved | |
| def _validate_value(self, name: str, value: Any, definition: ParameterDefinition) -> None: | |
| expected = definition.type | |
| valid = { | |
| ParameterType.STRING: isinstance(value, str), | |
| ParameterType.INTEGER: isinstance(value, int) and not isinstance(value, bool), | |
| ParameterType.NUMBER: isinstance(value, (int, float)) and not isinstance(value, bool), | |
| ParameterType.BOOLEAN: isinstance(value, bool), | |
| ParameterType.ARRAY: isinstance(value, list), | |
| ParameterType.OBJECT: isinstance(value, dict), | |
| }[expected] | |
| if not valid: | |
| raise TemplateValidationError(f"Template parameter '{name}' must be {expected.value}") | |
| if definition.enum is not None and value not in definition.enum: | |
| raise TemplateValidationError( | |
| f"Template parameter '{name}' must be one of the declared enum values" | |
| ) | |
| if isinstance(value, (int, float)) and not isinstance(value, bool): | |
| if definition.minimum is not None and value < definition.minimum: | |
| raise TemplateValidationError(f"Template parameter '{name}' is below its minimum") | |
| if definition.maximum is not None and value > definition.maximum: | |
| raise TemplateValidationError(f"Template parameter '{name}' exceeds its maximum") | |
| if isinstance(value, (str, list, dict)): | |
| if definition.min_length is not None and len(value) < definition.min_length: | |
| raise TemplateValidationError( | |
| f"Template parameter '{name}' is shorter than allowed" | |
| ) | |
| if definition.max_length is not None and len(value) > definition.max_length: | |
| raise TemplateValidationError(f"Template parameter '{name}' is longer than allowed") | |
| def _validate_variables( | |
| self, | |
| variables: set[str], | |
| known_parameters: set[str], | |
| template_id: str, | |
| step: int, | |
| ) -> None: | |
| for variable in variables: | |
| if variable not in known_parameters: | |
| raise TemplateValidationError( | |
| f"Template '{template_id}' references unknown parameter '{variable}'", | |
| details={"step": step}, | |
| ) | |
| def _variables(self, value: Any) -> set[str]: | |
| if isinstance(value, str): | |
| remainder = VARIABLE_PATTERN.sub("", value) | |
| if "{{" in remainder or "}}" in remainder: | |
| raise TemplateValidationError( | |
| "Template variables must use the exact '{{ parameter }}' syntax" | |
| ) | |
| return set(VARIABLE_PATTERN.findall(value)) | |
| if isinstance(value, dict): | |
| return ( | |
| set().union(*(self._variables(item) for item in value.values())) if value else set() | |
| ) | |
| if isinstance(value, list): | |
| return set().union(*(self._variables(item) for item in value)) if value else set() | |
| return set() | |
| def _substitute(self, value: Any, parameters: Mapping[str, Any]) -> Any: | |
| if isinstance(value, str): | |
| exact = EXACT_VARIABLE_PATTERN.fullmatch(value) | |
| if exact: | |
| name = exact.group(1) | |
| if name not in parameters: | |
| raise TemplateValidationError( | |
| f"Template parameter '{name}' has no resolved value" | |
| ) | |
| return copy.deepcopy(parameters[name]) | |
| def replace(match: re.Match[str]) -> str: | |
| name = match.group(1) | |
| if name not in parameters: | |
| raise TemplateValidationError( | |
| f"Template parameter '{name}' has no resolved value" | |
| ) | |
| replacement = parameters[name] | |
| if isinstance(replacement, (dict, list)): | |
| raise TemplateValidationError( | |
| f"Structured parameter '{name}' cannot be embedded in a string" | |
| ) | |
| return str(replacement) | |
| return VARIABLE_PATTERN.sub(replace, value) | |
| if isinstance(value, dict): | |
| return {key: self._substitute(item, parameters) for key, item in value.items()} | |
| if isinstance(value, list): | |
| return [self._substitute(item, parameters) for item in value] | |
| return copy.deepcopy(value) | |