Spaces:
Running
Running
File size: 4,308 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 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | 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
|