LCORE-4742: Implement prompt store operations

src/models/database/prompts.py

@@ -3,10 +3,9 @@
 from datetime import datetime
 
 from sqlalchemy import JSON, DateTime, ForeignKey, String, Text, func
-from sqlalchemy.orm import Mapped, mapped_column, validates
+from sqlalchemy.orm import Mapped, mapped_column
 
 from models.database.base import Base
-from prompts.exceptions import PromptValidationError
 
 
 class Prompt(Base):  # pylint: disable=too-few-public-methods
@@ -50,11 +49,3 @@ class PromptVersion(Base):  # pylint: disable=too-few-public-methods
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now()
     )
-
-    @validates("variables")
-    def validate_unique_variables(self, _key: str, value: list[str]) -> list[str]:
-        """Ensure no duplicate variable names."""
-        if len(value) != len(set(value)):
-            raise PromptValidationError("The variables contain duplicate names")
-
-        return value

src/prompts/exceptions.py

@@ -7,3 +7,7 @@ class PromptError(Exception):
 
 class PromptValidationError(PromptError):
     """Invalid prompt field values."""
+
+
+class PromptNotFoundError(PromptError):
+    """No prompt exists for the given identifier."""

src/prompts/store.py

@@ -0,0 +1,201 @@
+"""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, PromptValidationError
+
+logger = get_logger(__name__)
+
+
+def _validate_variables(variables: list[str]) -> None:
+    """Reject duplicate variable names.
+
+    Parameters:
+        variables: Declared variable names to validate.
+
+    Raises:
+        PromptValidationError: If the list contains duplicate names.
+
+    """
+    if len(variables) != len(set(variables)):
+        raise PromptValidationError("The variables contain duplicate names")
+
+
+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 variables contain duplicates.
+
+    """
+    # TODO: template validation  # pylint: disable=fixme
+
+    _validate_variables(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 variables contain duplicates.
+
+    """
+    # TODO: validate template  # pylint: disable=fixme
+
+    _validate_variables(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)

tests/unit/models/database/test_prompts.py

@@ -11,7 +11,6 @@
 
 from models.database.base import Base
 from models.database.prompts import Prompt, PromptVersion
-from prompts.exceptions import PromptValidationError
 
 
 @pytest.fixture(name="sqlite_engine")
@@ -265,47 +264,3 @@ def test_empty_variables_defaults_to_list(
         result = db_session.get(PromptVersion, (sample_prompt.id, 1))
         assert result is not None
         assert result.variables == []
-
-
-class TestPromptVersionVariableValidation:
-    """Test cases for PromptVersion variable uniqueness validation."""
-
-    def test_unique_variables_accepted(self) -> None:
-        """Test that a list of unique variable names is accepted."""
-        version = PromptVersion(
-            prompt_id="pmpt_001",
-            version=1,
-            template="{{a}} {{b}}",
-            variables=["a", "b"],
-        )
-        assert version.variables == ["a", "b"]
-
-    def test_duplicate_variables_raises_validation_error(self) -> None:
-        """Test that duplicate variable names raise PromptValidationError."""
-        with pytest.raises(PromptValidationError, match="duplicate names"):
-            PromptVersion(
-                prompt_id="pmpt_001",
-                version=1,
-                template="{{a}} {{a}}",
-                variables=["a", "a"],
-            )
-
-    def test_empty_variables_accepted(self) -> None:
-        """Test that an empty variable list is accepted."""
-        version = PromptVersion(
-            prompt_id="pmpt_001",
-            version=1,
-            template="No variables",
-            variables=[],
-        )
-        assert version.variables == []
-
-    def test_single_variable_accepted(self) -> None:
-        """Test that a single variable is accepted."""
-        version = PromptVersion(
-            prompt_id="pmpt_001",
-            version=1,
-            template="{{name}}",
-            variables=["name"],
-        )
-        assert version.variables == ["name"]

tests/unit/prompts/__init__.py

@@ -0,0 +1 @@
+"""Unit tests for the prompts module."""