Spaces:
Sleeping
Sleeping
File size: 1,604 Bytes
fba6023 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | 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
@dataclass(frozen=True, slots=True)
class TemplateReference:
"""Parsed `template_id@version` or `template_id@latest` reference."""
template_id: str
version: int | None
@classmethod
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
@dataclass(frozen=True, slots=True)
class PreparedTemplate:
"""Immutable runtime-ready template and resolved parameter values."""
definition: TemplateDefinition
parameters: dict[str, Any]
pipeline: tuple[ResolvedPipelineStep, ...]
output: OutputDefinition
|