LCORE-4743: Add Mustache-style template validation and rendering

src/models/database/__init__.py

@@ -1 +1,5 @@
 """Database models package."""
+
+from models.database.prompts import Prompt, PromptVersion
+
+__all__ = ["Prompt", "PromptVersion"]

src/models/database/prompts.py

@@ -0,0 +1,51 @@
+"""Persistence models for the versioned prompt catalog."""
+
+from datetime import datetime
+
+from sqlalchemy import JSON, DateTime, ForeignKey, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from models.database.base import Base
+
+
+class Prompt(Base):  # pylint: disable=too-few-public-methods
+    """Prompt identity with a pointer to the default version."""
+
+    __tablename__ = "prompt"
+
+    id: Mapped[str] = mapped_column(primary_key=True)
+
+    name: Mapped[str] = mapped_column(String(255))
+
+    default_version: Mapped[int] = mapped_column()
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+
+
+class PromptVersion(Base):  # pylint: disable=too-few-public-methods
+    """Immutable version row containing the template text and declared variables."""
+
+    __tablename__ = "prompt_version"
+
+    # Foreign key is not turned on by default for sqlite, but a PromptVersion will always
+    # be created after a Prompt is created in the database operation logic. Therefore,
+    # a PromptVersion with a nonexistent prompt.id will never be created.
+    prompt_id: Mapped[str] = mapped_column(
+        ForeignKey("prompt.id", ondelete="CASCADE"), primary_key=True
+    )
+
+    version: Mapped[int] = mapped_column(primary_key=True)
+
+    template: Mapped[str] = mapped_column(Text())
+
+    variables: Mapped[list[str]] = mapped_column(JSON, default=list)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )

src/prompts/__init__.py

@@ -0,0 +1 @@
+"""Prompts module for the versioned prompt catalog."""

src/prompts/exceptions.py

@@ -0,0 +1,17 @@
+"""Domain exceptions for the prompts module."""
+
+
+class PromptError(Exception):
+    """Base class for prompt domain errors."""
+
+
+class PromptValidationError(PromptError):
+    """Invalid prompt field values."""
+
+
+class PromptNotFoundError(PromptError):
+    """No prompt exists for the given identifier."""
+
+
+class PromptRenderError(PromptError):
+    """Prompt rendering error"""

src/prompts/store.py

@@ -0,0 +1,188 @@
+"""Store operations for the versioned prompt catalog."""
+
+from typing import Optional
+from uuid import uuid4
+
+from app.database import get_session
+from log import get_logger
+from models.database.prompts import Prompt, PromptVersion
+from prompts.exceptions import PromptNotFoundError
+from prompts.template import validate_template
+
+logger = get_logger(__name__)
+
+
+def create_prompt(name: str, template: str, variables: list[str]) -> PromptVersion:
+    """Create a new prompt with its initial version.
+
+    Parameters:
+        name: Display name for the prompt.
+        template: Mustache template text.
+        variables: Declared variable names for the template.
+
+    Returns:
+        PromptVersion: The initial version row.
+
+    Raises:
+        PromptValidationError: If the template is empty, contains invalid
+            placeholder names, has duplicate or invalid variable names,
+            or has undeclared placeholders.
+
+    """
+    validate_template(template, variables)
+
+    prompt_id = f"pmpt_{uuid4()}"
+    version = 1
+
+    with get_session() as session:
+        prompt = Prompt(id=prompt_id, name=name, default_version=version)
+        prompt_version = PromptVersion(
+            prompt_id=prompt_id, version=version, template=template, variables=variables
+        )
+
+        session.add(prompt)
+        session.add(prompt_version)
+        session.commit()
+        logger.debug("Created prompt id=%s with version %s", prompt_id, version)
+        session.refresh(prompt_version)
+
+        return prompt_version
+
+
+def get_prompt(prompt_id: str, version: Optional[int] = None) -> PromptVersion:
+    """Retrieve a prompt version by identifier.
+
+    When version is None the prompt's default version is used.
+
+    Parameters:
+        prompt_id: The prompt identifier.
+        version: Specific version number, or None for the default.
+
+    Returns:
+        PromptVersion: The matching version row.
+
+    Raises:
+        PromptNotFoundError: If the prompt or requested version does not exist.
+
+    """
+    with get_session() as session:
+
+        # Grab default version from Prompt if version is not provided
+        if version is None:
+            prompt = session.query(Prompt).filter_by(id=prompt_id).first()
+            if prompt is None:
+                logger.debug("Prompt id=%s not found", prompt_id)
+                raise PromptNotFoundError(f"Prompt id={prompt_id} not found")
+
+            version = prompt.default_version
+
+        prompt_version = (
+            session.query(PromptVersion)
+            .filter_by(prompt_id=prompt_id, version=version)
+            .first()
+        )
+
+        if prompt_version is None:
+            logger.debug("Version %s of prompt id=%s not found", version, prompt_id)
+            raise PromptNotFoundError(
+                f"Version {version} of prompt id={prompt_id} not found"
+            )
+
+        return prompt_version
+
+
+def list_prompts() -> list[Prompt]:
+    """List all prompts in the catalog.
+
+    Returns:
+        list[Prompt]: All prompt identity rows.
+
+    """
+    with get_session() as session:
+        prompts = session.query(Prompt).all()
+        return prompts
+
+
+def update_prompt(
+    prompt_id: str, template: str, variables: list[str], *, default: bool = False
+) -> PromptVersion:
+    """Create a new immutable version for an existing prompt.
+
+    Parameters:
+        prompt_id: The prompt identifier.
+        template: Mustache template text for the new version.
+        variables: Declared variable names for the template.
+        default: If True, set the new version as the prompt's default.
+
+    Returns:
+        PromptVersion: The newly created version row.
+
+    Raises:
+        PromptNotFoundError: If the prompt does not exist.
+        PromptValidationError: If the template is empty, contains invalid
+            placeholder names, has duplicate or invalid variable names,
+            or has undeclared placeholders.
+
+    """
+    validate_template(template, variables)
+
+    with get_session() as session:
+        max_prompt_version = (
+            session.query(PromptVersion)
+            .filter_by(prompt_id=prompt_id)
+            .order_by(PromptVersion.version.desc())
+            .first()
+        )
+        if max_prompt_version is None:
+            logger.debug("No available version found for prompt id=%s", prompt_id)
+            raise PromptNotFoundError(
+                f"No available version found for prompt id={prompt_id}"
+            )
+
+        max_version = max_prompt_version.version
+
+        prompt_version = PromptVersion(
+            prompt_id=prompt_id,
+            version=max_version + 1,
+            template=template,
+            variables=variables,
+        )
+        session.add(prompt_version)
+        logger.debug(
+            "Created version %s for prompt id=%s", prompt_version.version, prompt_id
+        )
+
+        if default:
+            prompt = session.query(Prompt).filter_by(id=prompt_id).first()
+            if prompt is None:
+                logger.debug("Prompt id=%s not found", prompt_id)
+                raise PromptNotFoundError(f"Prompt id={prompt_id} not found")
+
+            prompt.default_version = prompt_version.version
+
+        session.commit()
+        session.refresh(prompt_version)
+
+        return prompt_version
+
+
+def delete_prompt(prompt_id: str) -> None:
+    """Delete a prompt and all its versions.
+
+    Parameters:
+        prompt_id: The prompt identifier.
+
+    Raises:
+        PromptNotFoundError: If the prompt does not exist.
+
+    """
+    with get_session() as session:
+        prompt = session.query(Prompt).filter_by(id=prompt_id).first()
+        if prompt is None:
+            raise PromptNotFoundError(f"Prompt id={prompt_id} not found")
+
+        # This cover the situation when FK enforcement is not on
+        session.query(PromptVersion).filter_by(prompt_id=prompt_id).delete()
+        session.delete(prompt)
+        session.commit()
+        logger.debug("Deleted prompt id=%s", prompt_id)

src/prompts/template.py

@@ -0,0 +1,131 @@
+"""Mustache-style template validation and rendering.
+
+Placeholders use ``{{name}}`` syntax with optional inner whitespace.
+Only plain text substitution is supported; Jinja features (filters,
+control flow, expressions) are not allowed.
+
+Valid placeholder names must match ``[a-zA-Z_][a-zA-Z0-9_-]*``.
+Variables are text-only: each value is a plain string that replaces
+its corresponding placeholder verbatim.
+"""
+
+import re
+
+from models.database.prompts import PromptVersion
+from prompts.exceptions import PromptRenderError, PromptValidationError
+
+_PLACEHOLDER_RE = re.compile(r"\{\{\s*(.*?)\s*\}\}")
+
+_VALID_NAME_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_-]*$")
+
+
+def _validate_variables(variables: list[str]) -> set[str]:
+    """Validate variable names and reject duplicates.
+
+    Parameters:
+        variables: Declared variable names to validate.
+
+    Returns:
+        The validated variable names as a set.
+
+    Raises:
+        PromptValidationError: If the list contains duplicate names or
+            any name has an invalid format.
+
+    """
+    unique = set(variables)
+    if len(variables) != len(unique):
+        raise PromptValidationError("The variables contain duplicate names")
+
+    for variable in unique:
+        if not variable or not _VALID_NAME_RE.match(variable):
+            raise PromptValidationError(
+                f"The variable list contains invalid name format: {variable}"
+            )
+
+    return unique
+
+
+def _extract_placeholders(template: str) -> set[str]:
+    """Extract and validate all ``{{name}}`` placeholders from template text.
+
+    Parameters:
+        template: The template string to scan.
+
+    Returns:
+        Set of valid placeholder names found in the template.
+
+    Raises:
+        PromptValidationError: If any placeholder has an invalid name format.
+
+    """
+    raw_names = _PLACEHOLDER_RE.findall(template)
+    valid = set()
+    for raw_name in raw_names:
+        if not raw_name or not _VALID_NAME_RE.match(raw_name):
+            raise PromptValidationError(
+                f"The template contains invalid name format for the placeholder: {raw_name}"
+            )
+        valid.add(raw_name)
+    return valid
+
+
+def validate_template(template: str, variables: list[str]) -> None:
+    """Validate a template and its declared variables on create or update.
+
+    Every placeholder in the template must appear in the declared variables
+    list. Unused declared variables are allowed.
+
+    Parameters:
+        template: The template text to validate.
+        variables: Declared variable names for the template.
+
+    Raises:
+        PromptValidationError: If the template is empty, contains invalid
+            placeholder names, has duplicate variables, or has undeclared
+            placeholders.
+
+    """
+    if not template.strip():
+        raise PromptValidationError("The template must not be empty")
+
+    declared_variables = _validate_variables(variables)
+    placeholders = _extract_placeholders(template)
+    undeclared = placeholders - declared_variables
+
+    if undeclared:
+        raise PromptValidationError(f"Undeclared placeholders: {undeclared}")
+
+
+def render_template(prompt_version: PromptVersion, variables: dict[str, str]) -> str:
+    """Render a prompt version by substituting placeholder values.
+
+    Requires a value for every declared variable and rejects any extra keys.
+    The stored PromptVersion is not modified.
+
+    Parameters:
+        prompt_version: The validated prompt version to render.
+        variables: Mapping of variable names to their text substitution values.
+
+    Returns:
+        The rendered plain-text string.
+
+    Raises:
+        PromptRenderError: If required variables are missing or unexpected
+            variables are provided.
+
+    """
+    declared = set(prompt_version.variables)
+    provided = set(variables.keys())
+
+    missing = declared - provided
+    if missing:
+        raise PromptRenderError(f"Missing variables: {missing}")
+
+    extra = provided - declared
+    if extra:
+        raise PromptRenderError(f"Unexpected variables: {extra}")
+
+    return _PLACEHOLDER_RE.sub(
+        lambda m: variables[m.group(1).strip()], prompt_version.template
+    )