Feature/form analytics

tests/test_document_related.py

@@ -492,7 +492,7 @@ def test_historical_qty_document_per_corpus(self):
         test_hqdpc = HistoricalQtyDocumentPerCorpus(
             id=uuid.uuid4(),
             source_name="corpus_test",
-            quantity=100,
+            count=100,
         )
         test_session.add(test_hqdpc)
         test_session.commit()
@@ -513,7 +513,7 @@ def test_historical_qty_document_in_qdrant_per_corpus(self):
         test_hqdpc = HistoricalQtyDocumentInQdrantPerCorpus(
             id=uuid.uuid4(),
             source_name="corpus_test",
-            quantity=100,
+            count=100,
         )
         test_session.add(test_hqdpc)
         test_session.commit()
@@ -535,7 +535,7 @@ def test_historical_qty_document_in_qdrant(self):
 
         test_hqdpc = HistoricalQtyDocumentInQdrant(
             id=uuid.uuid4(),
-            quantity=100,
+            count=100,
         )
         test_session.add(test_hqdpc)
         test_session.commit()

tests/test_user_related.py

@@ -6,9 +6,10 @@
 from sqlalchemy.orm import sessionmaker
 
 from tests.helpers import handle_schema_with_sqlite
-from welearn_database.data.enumeration import FilterType
+from welearn_database.data.enumeration import FilterType, QuestionType
 from welearn_database.data.models import Base
 from welearn_database.data.models.user_related import (
+    AnalyticForm,
     APIKeyManagement,
     Bookmark,
     ChatMessage,
@@ -22,7 +23,6 @@
 from welearn_database.data.models.user_related import (
     UserProfile,
 )
-from welearn_database.exceptions import EarlyEnumerationVerificationError
 
 
 class TestUserRelatedCRUD(TestCase):
@@ -243,3 +243,71 @@ def test_create_and_read_filter_used_in_query(self):
             self.session.query(FilterUsedInQuery).filter_by(filter_value="13").first()
         )
         self.assertIsNotNone(result)
+
+    def test_create_and_read_analytic_form(self):
+        inferred_user = InferredUser(id=uuid.uuid4())
+        self.session.add(inferred_user)
+        self.session.commit()
+
+        user_session = UserSession(
+            id=uuid.uuid4(),
+            inferred_user_id=inferred_user.id,
+            origin_referrer="ref",
+            end_at=datetime.now() + timedelta(hours=1),
+            host="localhost",
+        )
+        self.session.add(user_session)
+        self.session.commit()
+
+        analytic_form = AnalyticForm(
+            id=uuid.uuid4(),
+            form_name="feedback_form",
+            question="Comment evaluez-vous la pertinence ?",
+            answer="Tres pertinente",
+            answer_type=QuestionType.TEXT.value,
+            session_id=user_session.id,
+        )
+        self.session.add(analytic_form)
+        self.session.commit()
+
+        result = (
+            self.session.query(AnalyticForm)
+            .filter_by(form_name="feedback_form", session_id=user_session.id)
+            .first()
+        )
+        self.assertIsNotNone(result)
+        self.assertEqual(result.question, "Comment evaluez-vous la pertinence ?")
+        self.assertEqual(result.answer, "Tres pertinente")
+        self.assertEqual(result.answer_type, QuestionType.TEXT.value)
+        self.assertEqual(result.session_id, user_session.id)
+
+    def test_analytic_form_session_relationship(self):
+        inferred_user = InferredUser(id=uuid.uuid4())
+        self.session.add(inferred_user)
+        self.session.commit()
+
+        user_session = UserSession(
+            id=uuid.uuid4(),
+            inferred_user_id=inferred_user.id,
+            origin_referrer="ref",
+            end_at=datetime.now() + timedelta(hours=1),
+            host="localhost",
+        )
+        self.session.add(user_session)
+        self.session.commit()
+
+        analytic_form = AnalyticForm(
+            id=uuid.uuid4(),
+            form_name="feedback_form",
+            question="Question test",
+            answer="Reponse test",
+            answer_type=QuestionType.PREDEFINED_TEXT.value,
+            session_id=user_session.id,
+        )
+        self.session.add(analytic_form)
+        self.session.commit()
+
+        result = self.session.query(AnalyticForm).filter_by(id=analytic_form.id).first()
+        self.assertIsNotNone(result)
+        self.assertIsNotNone(result.session)
+        self.assertEqual(result.session.id, user_session.id)

welearn_database/alembic/versions/f5e91a5c950d_add_analytic_form.py

@@ -0,0 +1,67 @@
+"""add analytic form
+
+Revision ID: f5e91a5c950d
+Revises: 6d4346fad6f4
+Create Date: 2026-09-02 16:32:55.188400
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "f5e91a5c950d"
+down_revision: Union[str, None] = "6d4346fad6f4"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "analytic_form",
+        sa.Column(
+            "id", sa.Uuid(), server_default=sa.func.gen_random_uuid(), nullable=False
+        ),
+        sa.Column("form_name", sa.String(), nullable=False),
+        sa.Column("question", sa.String(), nullable=False),
+        sa.Column("answer", sa.String(), nullable=False),
+        sa.Column(
+            "answer_type",
+            postgresql.ENUM(
+                "text",
+                "checkbox",
+                "predefined_text",
+                name="answer_type",
+                schema="user_related",
+            ),
+            nullable=False,
+        ),
+        sa.Column("session_id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=False),
+            nullable=False,
+            server_default="NOW()",
+        ),
+        sa.ForeignKeyConstraint(
+            ["session_id"],
+            ["user_related.session.id"],
+            name="analytic_form_session_id_fkey",
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        schema="user_related",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        "analytic_form_session_id_fkey",
+        "analytic_form",
+        schema="user_related",
+        type_="foreignkey",
+    )
+    op.drop_table("analytic_form", schema="user_related")
+    op.execute("DROP TYPE IF EXISTS user_related.answer_type")

welearn_database/data/enumeration.py

@@ -43,3 +43,9 @@ class ExternalIdType(StrEnum):
 class FilterType(StrEnum):
     SDG = auto()
     SOURCE = auto()
+
+
+class QuestionType(StrEnum):
+    TEXT = auto()
+    CHECKBOX = auto()
+    PREDEFINED_TEXT = auto()

welearn_database/data/models/user_related.py

@@ -5,7 +5,7 @@
 from sqlalchemy.dialects.postgresql import ENUM, TIMESTAMP
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
-from welearn_database.data.enumeration import DbSchemaEnum, FilterType
+from welearn_database.data.enumeration import DbSchemaEnum, FilterType, QuestionType
 from welearn_database.data.models.document_related import WeLearnDocument
 
 from . import Base
@@ -271,3 +271,35 @@ class FilterUsedInQuery(Base):
     filter_value: Mapped[str]
 
     chat_message: Mapped["ChatMessage"] = relationship()
+
+
+class AnalyticForm(Base):
+    __tablename__ = "analytic_form"
+    __table_args__ = {"schema": DbSchemaEnum.USER_RELATED.value}
+
+    id: Mapped[UUID] = mapped_column(
+        types.Uuid, primary_key=True, nullable=False, server_default="gen_random_uuid()"
+    )
+    form_name: Mapped[str] = mapped_column(nullable=False)
+    question: Mapped[str] = mapped_column(nullable=False)
+    answer: Mapped[str] = mapped_column(nullable=False)
+    answer_type: Mapped[str] = mapped_column(
+        ENUM(
+            *(e.value.lower() for e in QuestionType),
+            name="answer_type",
+            schema=DbSchemaEnum.USER_RELATED.value,
+        ),
+        nullable=False,
+    )
+    session_id: Mapped[UUID] = mapped_column(
+        types.Uuid,
+        ForeignKey(f"{DbSchemaEnum.USER_RELATED.value}.session.id"),
+        nullable=False,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=False),
+        nullable=False,
+        default=func.localtimestamp(),
+        server_default="NOW()",
+    )
+    session = relationship("Session", foreign_keys=[session_id])