Spaces:
Running
Running
| from __future__ import annotations | |
| from pathlib import Path | |
| from typing import Any | |
| import yaml | |
| from pydantic import ValidationError | |
| from app.core.exceptions import TemplateValidationError | |
| from app.core.logger import get_logger | |
| from app.templates.schema import TemplateDefinition | |
| from app.templates.validator import TemplateValidator | |
| logger = get_logger(__name__) | |
| class TemplateLoader: | |
| """Recursively loads and validates safe YAML template documents.""" | |
| MAX_TEMPLATE_BYTES = 1_048_576 | |
| def __init__(self, root: Path, validator: TemplateValidator) -> None: | |
| self.root = root.resolve() | |
| self.validator = validator | |
| def load(self) -> list[TemplateDefinition]: | |
| """Scan the configured directory and return every valid definition.""" | |
| if not self.root.is_dir(): | |
| raise TemplateValidationError( | |
| "Template directory does not exist", details={"path": str(self.root)} | |
| ) | |
| definitions: list[TemplateDefinition] = [] | |
| try: | |
| paths = sorted((*self.root.rglob("*.yaml"), *self.root.rglob("*.yml"))) | |
| except OSError as exc: | |
| raise TemplateValidationError( | |
| "Template directory could not be scanned", | |
| details={"path": str(self.root)}, | |
| ) from exc | |
| for path in paths: | |
| definitions.extend(self._load_file(path)) | |
| if not definitions: | |
| raise TemplateValidationError( | |
| "No YAML templates were found", details={"path": str(self.root)} | |
| ) | |
| logger.info( | |
| "templates loaded", | |
| extra={"path": str(self.root), "templates": len(definitions)}, | |
| ) | |
| return definitions | |
| def _load_file(self, path: Path) -> list[TemplateDefinition]: | |
| resolved_path = path.resolve() | |
| if self.root not in resolved_path.parents: | |
| raise TemplateValidationError( | |
| "Template YAML path escapes TEMPLATE_DIR", | |
| details={"path": str(path)}, | |
| ) | |
| try: | |
| size = resolved_path.stat().st_size | |
| except OSError as exc: | |
| raise TemplateValidationError( | |
| "Template YAML file could not be inspected", | |
| details={"path": str(path)}, | |
| ) from exc | |
| if size > self.MAX_TEMPLATE_BYTES: | |
| raise TemplateValidationError( | |
| "Template YAML file is too large", details={"path": str(path)} | |
| ) | |
| try: | |
| documents = list(yaml.safe_load_all(resolved_path.read_text(encoding="utf-8"))) | |
| except (OSError, UnicodeError, yaml.YAMLError) as exc: | |
| raise TemplateValidationError( | |
| "Template YAML syntax is invalid", details={"path": str(path)} | |
| ) from exc | |
| raw_templates: list[Any] = [] | |
| for document in documents: | |
| if document is None: | |
| continue | |
| if isinstance(document, dict) and set(document) == {"templates"}: | |
| collection = document["templates"] | |
| if not isinstance(collection, list): | |
| raise TemplateValidationError( | |
| "The templates YAML key must contain a list", | |
| details={"path": str(path)}, | |
| ) | |
| raw_templates.extend(collection) | |
| elif isinstance(document, list): | |
| raw_templates.extend(document) | |
| else: | |
| raw_templates.append(document) | |
| definitions: list[TemplateDefinition] = [] | |
| for index, raw in enumerate(raw_templates): | |
| try: | |
| definition = TemplateDefinition.model_validate(raw) | |
| self.validator.validate_definition(definition) | |
| except ValidationError as exc: | |
| raise TemplateValidationError( | |
| "Template YAML does not match the schema", | |
| details={ | |
| "path": str(path), | |
| "document": index, | |
| "errors": exc.errors( | |
| include_url=False, include_context=False, include_input=False | |
| ), | |
| }, | |
| ) from exc | |
| definitions.append(definition) | |
| return definitions | |