From 28701d87f339cf4a4a654158b8003256ca4419f1 Mon Sep 17 00:00:00 2001 From: Jazzcort Date: Wed, 28 Oct 2026 15:45:31 -0400 Subject: [PATCH 1/2] LCORE-4741: Add prompts module skeleton and persistence models Introduce Prompt and PromptVersion ORM models for the local prompt catalog. Prompt holds identity and a default-version pointer; PromptVersion uses a (prompt_id, version) composite primary key for immutable version rows with template text and declared variables. Includes the prompts module skeleton with domain exceptions and unit tests covering schema definition, constraints, and variable validation. --- src/models/database/prompts.py | 58 ++++ src/prompts/__init__.py | 1 + src/prompts/exceptions.py | 9 + tests/unit/models/database/test_prompts.py | 311 +++++++++++++++++++++ 4 files changed, 379 insertions(+) create mode 100644 src/models/database/prompts.py create mode 100644 src/prompts/__init__.py create mode 100644 src/prompts/exceptions.py create mode 100644 tests/unit/models/database/test_prompts.py diff --git a/src/models/database/prompts.py b/src/models/database/prompts.py new file mode 100644 index 000000000..cf701acda --- /dev/null +++ b/src/models/database/prompts.py @@ -0,0 +1,58 @@ +"""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, validates + +from models.database.base import Base +from prompts.exceptions import PromptValidationError + + +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"), 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() + ) + + @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 diff --git a/src/prompts/__init__.py b/src/prompts/__init__.py new file mode 100644 index 000000000..14fcd39dd --- /dev/null +++ b/src/prompts/__init__.py @@ -0,0 +1 @@ +"""Prompts module for the versioned prompt catalog.""" diff --git a/src/prompts/exceptions.py b/src/prompts/exceptions.py new file mode 100644 index 000000000..8b23ce113 --- /dev/null +++ b/src/prompts/exceptions.py @@ -0,0 +1,9 @@ +"""Domain exceptions for the prompts module.""" + + +class PromptError(Exception): + """Base class for prompt domain errors.""" + + +class PromptValidationError(PromptError): + """Invalid prompt field values.""" diff --git a/tests/unit/models/database/test_prompts.py b/tests/unit/models/database/test_prompts.py new file mode 100644 index 000000000..6e998d20a --- /dev/null +++ b/tests/unit/models/database/test_prompts.py @@ -0,0 +1,311 @@ +"""Unit tests for Prompt and PromptVersion database models.""" + +from collections.abc import Generator + +import pytest +from sqlalchemy import String, create_engine, inspect, text +from sqlalchemy.engine import Engine +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from models.database.base import Base +from models.database.prompts import Prompt, PromptVersion +from prompts.exceptions import PromptValidationError + + +@pytest.fixture(name="sqlite_engine") +def sqlite_engine_fixture() -> Generator[Engine, None, None]: + """Provide an in-memory SQLite engine with all tables created. + + Yields: + Engine: A SQLAlchemy Engine bound to an in-memory SQLite database with + tables created from Base.metadata. + + """ + engine = create_engine( + "sqlite:///:memory:", + echo=False, + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(engine) + yield engine + engine.dispose() + + +@pytest.fixture(name="db_session") +def db_session_fixture(sqlite_engine: Engine) -> Generator[Session, None, None]: + """Provide a database session for tests. + + Parameters: + sqlite_engine: In-memory SQLite engine fixture. + + Yields: + Session: A SQLAlchemy Session bound to the test engine. + + """ + session_factory = sessionmaker( + autocommit=False, autoflush=False, bind=sqlite_engine + ) + session = session_factory() + try: + yield session + finally: + session.close() + + +@pytest.fixture(name="sample_prompt") +def sample_prompt_fixture(db_session: Session) -> Prompt: + """Insert and return a sample Prompt row. + + Parameters: + db_session: Active database session. + + Returns: + Prompt: The persisted prompt instance. + + """ + prompt = Prompt(id="pmpt_001", name="greeting", default_version=1) + db_session.add(prompt) + db_session.commit() + return prompt + + +class TestPromptModelDefinition: + """Test cases for Prompt model schema definition.""" + + def test_tablename(self) -> None: + """Test Prompt uses the expected table name.""" + assert Prompt.__tablename__ == "prompt" + + def test_expected_columns_exist(self) -> None: + """Test Prompt defines all expected columns.""" + column_names = {col.name for col in Prompt.__table__.columns} + expected = {"id", "name", "default_version", "created_at", "updated_at"} + assert column_names == expected + + def test_name_column_max_length(self) -> None: + """Test name column has max length 255.""" + name_column = Prompt.__table__.c.name + assert isinstance(name_column.type, String) + assert name_column.type.length == 255 + + def test_id_is_primary_key(self) -> None: + """Test id column is the primary key.""" + assert Prompt.__table__.c.id.primary_key + + +class TestPromptVersionModelDefinition: + """Test cases for PromptVersion model schema definition.""" + + def test_tablename(self) -> None: + """Test PromptVersion uses the expected table name.""" + assert PromptVersion.__tablename__ == "prompt_version" + + def test_expected_columns_exist(self) -> None: + """Test PromptVersion defines all expected columns.""" + column_names = {col.name for col in PromptVersion.__table__.columns} + expected = {"prompt_id", "version", "template", "variables", "created_at"} + assert column_names == expected + + def test_composite_primary_key(self) -> None: + """Test primary key is (prompt_id, version).""" + pk_columns = {col.name for col in PromptVersion.__table__.primary_key.columns} + assert pk_columns == {"prompt_id", "version"} + + def test_prompt_id_has_foreign_key(self) -> None: + """Test prompt_id references prompt.id.""" + fk_targets = { + fk.target_fullname + for fk in PromptVersion.__table__.c.prompt_id.foreign_keys + } + assert "prompt.id" in fk_targets + + +class TestPromptDatabaseOperations: + """Test cases for Prompt database operations.""" + + def test_tables_created_by_metadata(self, sqlite_engine: Engine) -> None: + """Test prompt and prompt_version tables are created via Base.metadata.""" + inspector = inspect(sqlite_engine) + table_names = inspector.get_table_names() + assert "prompt" in table_names + assert "prompt_version" in table_names + + def test_insert_and_read_prompt(self, db_session: Session) -> None: + """Test inserting and reading back a Prompt row.""" + prompt = Prompt(id="pmpt_001", name="greeting", default_version=1) + db_session.add(prompt) + db_session.commit() + + result = db_session.get(Prompt, "pmpt_001") + assert result is not None + assert result.id == "pmpt_001" + assert result.name == "greeting" + assert result.default_version == 1 + assert result.created_at is not None + assert result.updated_at is not None + + def test_duplicate_prompt_id_raises_integrity_error( + self, db_session: Session + ) -> None: + """Test inserting duplicate prompt id raises IntegrityError.""" + db_session.add(Prompt(id="pmpt_001", name="first", default_version=1)) + db_session.commit() + + db_session.add(Prompt(id="pmpt_001", name="second", default_version=1)) + with pytest.raises(IntegrityError): + db_session.commit() + + +class TestPromptVersionDatabaseOperations: + """Test cases for PromptVersion database operations.""" + + def test_insert_and_read_version( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test inserting and reading back a PromptVersion row.""" + version = PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="Hello {{name}}", + variables=["name"], + ) + db_session.add(version) + db_session.commit() + + result = db_session.get(PromptVersion, (sample_prompt.id, 1)) + assert result is not None + assert result.prompt_id == sample_prompt.id + assert result.version == 1 + assert result.template == "Hello {{name}}" + assert result.variables == ["name"] + assert result.created_at is not None + + def test_multiple_versions_for_same_prompt( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test that a prompt can have multiple versions.""" + v1 = PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="Hello {{name}}", + variables=["name"], + ) + v2 = PromptVersion( + prompt_id=sample_prompt.id, + version=2, + template="Hi {{name}}, welcome to {{place}}", + variables=["name", "place"], + ) + db_session.add_all([v1, v2]) + db_session.commit() + + results = ( + db_session.query(PromptVersion) + .filter_by(prompt_id=sample_prompt.id) + .order_by(PromptVersion.version) + .all() + ) + assert len(results) == 2 + assert results[0].version == 1 + assert results[1].version == 2 + + def test_duplicate_prompt_id_and_version_raises_integrity_error( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test inserting duplicate (prompt_id, version) raises IntegrityError.""" + db_session.add( + PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="first", + variables=[], + ) + ) + db_session.commit() + + db_session.add( + PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="second", + variables=[], + ) + ) + with pytest.raises(IntegrityError): + db_session.commit() + + def test_foreign_key_enforced(self, db_session: Session) -> None: + """Test inserting a version for a non-existent prompt raises IntegrityError.""" + db_session.execute(text("PRAGMA foreign_keys = ON")) + version = PromptVersion( + prompt_id="nonexistent", + version=1, + template="Hello", + variables=[], + ) + db_session.add(version) + with pytest.raises(IntegrityError): + db_session.commit() + + def test_empty_variables_defaults_to_list( + self, db_session: Session, sample_prompt: Prompt + ) -> None: + """Test that variables defaults to an empty list.""" + version = PromptVersion( + prompt_id=sample_prompt.id, + version=1, + template="No placeholders here", + ) + db_session.add(version) + db_session.commit() + + 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"] From 280e2e9721eb982d11ad6bb3b195c06cc23738b8 Mon Sep 17 00:00:00 2001 From: Jazzcort Date: Thu, 29 Oct 2026 09:39:26 -0400 Subject: [PATCH 2/2] Import Prompt and PromptVersion at __init__.py This makes sure both tables are registered when we import Base at src/app/database.py. All the other tables (i.e. SavedPrompt, UserConversation, and UserTurn) are registered transitively, so it will be better we also import those tables at __init__.py. --- src/models/database/__init__.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/models/database/__init__.py b/src/models/database/__init__.py index a01a86e5c..9a4d012fb 100644 --- a/src/models/database/__init__.py +++ b/src/models/database/__init__.py @@ -1 +1,5 @@ """Database models package.""" + +from models.database.prompts import Prompt, PromptVersion + +__all__ = ["Prompt", "PromptVersion"]