"""Template library — stack-shaped folders. A template is just a directory laid out like a real stack (``compose.yaml`` plus optional ``.env.example`` and any extra files), accompanied by a small ``template.json`` describing it. "Pulling" a template copies the whole folder into a new stack, which is then editable like any other stack. Two roots are scanned: * **bundled** — ``backend/templates/`` ships in the image / git repo (read-only). * **custom** — ``${DATA_DIR}/templates/`` is writable and persists on the data volume; this is where "save stack as template" writes to. """ from __future__ import annotations import json import os import re import shutil from typing import Optional from config import settings from services import compose_service BUNDLED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "templates") _META_NAME = "template.json" _ENV_EXAMPLE = ".env.example" _CUSTOM_PREFIX = "custom:" def custom_dir() -> str: return os.path.join(settings.DATA_DIR, "templates") # --------------------------------------------------------------------------- # # Low-level helpers # --------------------------------------------------------------------------- # def _read_file(path: str) -> str: try: with open(path, "r", encoding="utf-8") as fh: return fh.read() except OSError: return "" def _load_meta(folder: str, slug: str) -> dict: """Metadata from template.json, with sensible fallbacks.""" meta: dict = {} raw = _read_file(os.path.join(folder, _META_NAME)) if raw: try: meta = json.loads(raw) except json.JSONDecodeError: meta = {} return { "name": meta.get("name") or slug.replace("-", " ").title(), "description": meta.get("description"), "tags": meta.get("tags") or [], "gpu": meta.get("gpu"), } def _is_template(folder: str) -> bool: return os.path.isdir(folder) and compose_service.find_compose_file(folder) is not None def _list_files(folder: str) -> list[str]: """Relative paths shipped by the template (excludes the metadata file).""" out: list[str] = [] for root, _dirs, files in os.walk(folder): for fn in sorted(files): rel = os.path.relpath(os.path.join(root, fn), folder) if rel == _META_NAME: continue out.append(rel) return sorted(out) def _resolve_dir(template_id: str) -> Optional[str]: """Map a template id to its on-disk folder (guards against traversal).""" if template_id.startswith(_CUSTOM_PREFIX): slug = template_id[len(_CUSTOM_PREFIX):] base = custom_dir() else: slug = template_id base = BUNDLED_DIR slug = os.path.basename(slug.strip()) if not slug: return None path = os.path.join(base, slug) return path if _is_template(path) else None def _scan(base: str, source: str) -> list[dict]: out: list[dict] = [] if not os.path.isdir(base): return out for slug in sorted(os.listdir(base)): folder = os.path.join(base, slug) if not _is_template(folder): continue meta = _load_meta(folder, slug) out.append( { "id": f"{_CUSTOM_PREFIX}{slug}" if source == "custom" else slug, "name": meta["name"], "description": meta["description"], "tags": meta["tags"], "gpu": meta["gpu"], "source": source, } ) return out # --------------------------------------------------------------------------- # # Listing / detail # --------------------------------------------------------------------------- # def list_templates() -> list[dict]: return _scan(BUNDLED_DIR, "bundled") + _scan(custom_dir(), "custom") def get_template(template_id: str) -> Optional[dict]: folder = _resolve_dir(template_id) if not folder: return None slug = os.path.basename(folder) meta = _load_meta(folder, slug) compose_file = compose_service.find_compose_file(folder) return { "id": template_id, "name": meta["name"], "description": meta["description"], "tags": meta["tags"], "gpu": meta["gpu"], "source": "custom" if template_id.startswith(_CUSTOM_PREFIX) else "bundled", "compose": _read_file(compose_file) if compose_file else "", "env": _read_file(os.path.join(folder, _ENV_EXAMPLE)), "files": _list_files(folder), } # --------------------------------------------------------------------------- # # Pull (instantiate) — copy the whole folder into a new stack # --------------------------------------------------------------------------- # def copy_into_stack(template_id: str, stack_id: str, override: Optional[str] = None) -> None: """Copy a template folder into a fresh stack directory. The ``template.json`` is left behind and any ``.env.example`` is promoted to a real ``.env`` so the pulled stack is immediately runnable + editable. """ src = _resolve_dir(template_id) if not src: raise FileNotFoundError(f"Template '{template_id}' not found") dst = compose_service.stack_dir(stack_id, override) if os.path.exists(dst): raise FileExistsError(f"Stack '{stack_id}' already exists") shutil.copytree(src, dst, ignore=shutil.ignore_patterns(_META_NAME)) example = os.path.join(dst, _ENV_EXAMPLE) env = os.path.join(dst, ".env") if os.path.isfile(example) and not os.path.isfile(env): os.replace(example, env) # --------------------------------------------------------------------------- # # Save / delete custom templates (folder-based, persisted on the data volume) # --------------------------------------------------------------------------- # def save_custom( name: str, compose: str, env: str = "", description: str = "", tags: list[str] | None = None, gpu: str | None = None, ) -> str: """Write a custom template folder; returns its slug. Overwrites if it exists.""" slug = compose_service.slugify(name) folder = os.path.join(custom_dir(), slug) os.makedirs(folder, exist_ok=True) meta = { "name": name, "description": description or None, "tags": tags or [], "gpu": gpu, } with open(os.path.join(folder, _META_NAME), "w", encoding="utf-8") as fh: json.dump(meta, fh, indent=2) fh.write("\n") with open(os.path.join(folder, compose_service.DEFAULT_COMPOSE_NAME), "w", encoding="utf-8") as fh: fh.write(compose or "services:\n") example = os.path.join(folder, _ENV_EXAMPLE) if env.strip(): with open(example, "w", encoding="utf-8") as fh: fh.write(env) elif os.path.isfile(example): os.remove(example) return slug def save_from_stack(stack_id: str, name: str, description: str = "") -> str: """Snapshot an existing stack's compose + env into a custom template.""" compose = compose_service.read_compose(stack_id) env = compose_service.read_env(stack_id) return save_custom(name, compose, env, description=description) def delete_custom(slug: str) -> bool: folder = os.path.join(custom_dir(), os.path.basename(slug.strip())) if not _is_template(folder): return False shutil.rmtree(folder) return True # --------------------------------------------------------------------------- # # Legacy migration (pre-0.31 custom templates lived in the database) # --------------------------------------------------------------------------- # _LEGACY_VAR_RE = re.compile(r"\{\{\s*([A-Za-z0-9_]+)\s*\}\}") def migrate_legacy_db_templates() -> int: """One-time: move custom templates out of the dropped ``template`` table. Old templates used ``{{VAR}}`` placeholders; compose interpolates ``${VAR}`` from ``.env``, so placeholders are rewritten and the variable names land in the template's ``.env.example``. Returns the number of templates moved. """ from sqlalchemy import inspect, text from database import engine if not inspect(engine).has_table("template"): return 0 moved = 0 with engine.begin() as conn: rows = conn.execute( text("SELECT name, description, tags, yaml FROM template") ).all() for name, description, tags, yaml_str in rows: compose = _LEGACY_VAR_RE.sub(r"${\1}", yaml_str or "") variables = dict.fromkeys(_LEGACY_VAR_RE.findall(yaml_str or "")) env = "".join(f"{v}=\n" for v in variables) save_custom( name or "template", compose, env, description=description or "", tags=[t for t in (tags or "").split(",") if t], ) moved += 1 conn.execute(text("DROP TABLE template")) return moved