from __future__ import annotations import re from enum import Enum from typing import Any from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator IDENTIFIER_PATTERN = r"^[a-z][a-z0-9_]*$" class ParameterType(str, Enum): """Supported runtime parameter types for YAML templates.""" STRING = "string" INTEGER = "integer" NUMBER = "number" BOOLEAN = "boolean" ARRAY = "array" OBJECT = "object" class ParameterDefinition(BaseModel): """Declarative validation contract for one runtime template parameter.""" model_config = ConfigDict(extra="forbid") type: ParameterType description: str = "" required: bool = Field(default=False, strict=True) default: Any = None enum: list[Any] | None = None minimum: float | None = None maximum: float | None = None min_length: int | None = Field(default=None, ge=0) max_length: int | None = Field(default=None, ge=0) @property def has_default(self) -> bool: """Return whether YAML explicitly supplied a default value.""" return "default" in self.model_fields_set @model_validator(mode="after") def validate_bounds(self) -> ParameterDefinition: if self.minimum is not None and self.maximum is not None: if self.minimum > self.maximum: raise ValueError("minimum must not exceed maximum") if self.min_length is not None and self.max_length is not None: if self.min_length > self.max_length: raise ValueError("min_length must not exceed max_length") return self class PipelineStep(BaseModel): """One operation invocation in a template pipeline.""" model_config = ConfigDict(extra="allow") operation: str = Field(pattern=IDENTIFIER_PATTERN) when: Any = True inputs: list[str] | None = None save_as: str | None = Field(default=None, pattern=IDENTIFIER_PATTERN) def operation_parameters(self) -> dict[str, Any]: """Return operation arguments declared as extra YAML keys.""" return dict(self.model_extra or {}) class OutputDefinition(BaseModel): """Declared output contract for a template.""" model_config = ConfigDict(extra="forbid") format: str filename: str | None = None class TemplateDefinition(BaseModel): """Validated, versioned YAML media workflow definition.""" model_config = ConfigDict(extra="forbid") id: str = Field(pattern=IDENTIFIER_PATTERN) name: str = Field(min_length=1, max_length=120) category: str = Field(pattern=IDENTIFIER_PATTERN) description: str = Field(min_length=1, max_length=1000) author: str = Field(min_length=1, max_length=120) version: int = Field(ge=1, strict=True) tags: list[str] = Field(min_length=1) estimated_runtime: str = Field(min_length=1, max_length=80) supported_inputs: list[str] = Field(min_length=1) supported_outputs: list[str] = Field(min_length=1) parameters: dict[str, ParameterDefinition] = Field(default_factory=dict) pipeline: list[PipelineStep] = Field(min_length=1) output: OutputDefinition examples: list[dict[str, Any]] = Field(default_factory=list) @field_validator("tags", "supported_inputs", "supported_outputs") @classmethod def normalize_string_list(cls, values: list[str]) -> list[str]: normalized = [value.strip().lower() for value in values if value.strip()] if not normalized: raise ValueError("list must contain at least one non-empty value") return list(dict.fromkeys(normalized)) @field_validator("parameters") @classmethod def validate_parameter_names( cls, values: dict[str, ParameterDefinition] ) -> dict[str, ParameterDefinition]: for name in values: if not re.fullmatch(IDENTIFIER_PATTERN, name): raise ValueError(f"invalid parameter name: {name}") return values