Spaces:
Running
Running
| from __future__ import annotations | |
| import re | |
| from dataclasses import dataclass | |
| from typing import Any | |
| from pydantic import BaseModel, ConfigDict, Field | |
| from app.core.exceptions import TemplateValidationError | |
| from app.templates.schema import OutputDefinition, TemplateDefinition | |
| class TemplateReference: | |
| """Parsed `template_id@version` or `template_id@latest` reference.""" | |
| template_id: str | |
| version: int | None | |
| def parse(cls, value: str) -> TemplateReference: | |
| normalized = value.strip().lower() | |
| match = re.fullmatch(r"([a-z][a-z0-9_]*)(?:@([1-9][0-9]*|latest))?", normalized) | |
| if not match: | |
| raise TemplateValidationError( | |
| "Template reference must use id, id@version, or id@latest" | |
| ) | |
| version_text = match.group(2) | |
| version = None if version_text in {None, "latest"} else int(version_text) | |
| return cls(template_id=match.group(1), version=version) | |
| class ResolvedPipelineStep(BaseModel): | |
| """Pipeline step after safe parameter substitution.""" | |
| model_config = ConfigDict(extra="forbid") | |
| operation: str | |
| parameters: dict[str, Any] = Field(default_factory=dict) | |
| enabled: bool = True | |
| inputs: list[str] | None = None | |
| save_as: str | None = None | |
| class PreparedTemplate: | |
| """Immutable runtime-ready template and resolved parameter values.""" | |
| definition: TemplateDefinition | |
| parameters: dict[str, Any] | |
| pipeline: tuple[ResolvedPipelineStep, ...] | |
| output: OutputDefinition | |