[feature/PI-565-questionnaire_rethink] add new spine questionnaires

Author: jaklinger

pyproject.toml

@@ -27,6 +27,7 @@ urllib3 = "<3"
 orjson = "^3.9.15"
 attrs = "^24.2.0"
 locust = "^2.29.1"
+jsonschema = "^4.23.0"
 
 [tool.poetry.group.dev.dependencies]
 pre-commit = "^4.0.0"

src/layers/domain/core/questionnaire/tests/test_questionnaire_v3.py

@@ -0,0 +1,93 @@
+import json
+
+import pytest
+from domain.core.enum import Status
+from domain.core.questionnaire.v3 import Questionnaire
+from domain.core.timestamp import now
+from jsonschema import ValidationError as JsonSchemaValidationError
+from pydantic import ValidationError as PydanticValidationError
+
+VALID_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+        "size": {
+            "type": "number",
+            "minimum": 1,
+            "maximum": 14,
+        },
+        "colour": {
+            "type": "string",
+            "enum": ["black", "white"],
+        },
+        "brand": {"type": "string"},  # not required
+    },
+    "required": ["size", "colour"],
+    "additionalProperties": False,
+}
+
+INVALID_SCHEMA = {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+        "a-field": {
+            "type": "not-a-type",
+        }
+    },
+    "required": ["a-field"],
+}
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        {"size": 1, "colour": "black"},
+        {"size": 14, "colour": "white"},
+        {"size": 7, "colour": "white", "brand": "something"},
+    ],
+)
+def test_schema_validation_pass(data):
+    questionnaire = Questionnaire(
+        name="foo", version="1", json_schema=json.dumps(VALID_SCHEMA)
+    )
+    response = questionnaire.validate(data=data)
+    assert response.name == "foo"
+    assert response.version == "1"
+    assert response.data == data
+    assert response.status is Status.ACTIVE
+    assert response.created_on.date() == now().date()
+    assert response.updated_on is None
+    assert response.deleted_on is None
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        {"size": 1, "colour": "red"},
+        {"size": "not a number", "colour": "white"},
+        {
+            "size": 7,
+            "colour": "white",
+            "brand": "something",
+            "unknown_field": "foo",
+        },
+    ],
+)
+def test_schema_validation_fail(data):
+    questionnaire = Questionnaire(
+        name="foo", version="1", json_schema=json.dumps(VALID_SCHEMA)
+    )
+    with pytest.raises(JsonSchemaValidationError):
+        questionnaire.validate(data=data)
+
+
+@pytest.mark.parametrize(
+    "schema",
+    [
+        {},
+        INVALID_SCHEMA,
+    ],
+)
+def test_invalid_schema(schema):
+    with pytest.raises(PydanticValidationError):
+        Questionnaire(name="name", version="123", json_schema=json.dumps(schema))

src/layers/domain/core/questionnaire/v3.py

@@ -0,0 +1,35 @@
+from datetime import datetime
+from uuid import UUID, uuid4
+
+import jsonschema
+from domain.core.base import BaseModel
+from domain.core.timestamp import now
+from pydantic import Field, Json, validator
+
+
+class Questionnaire(BaseModel):
+    name: str
+    version: str
+    json_schema: Json
+
+    @validator("json_schema")
+    def validate_json_schema(cls, json_schema):
+        try:
+            jsonschema.Draft7Validator.check_schema(json_schema)
+        except jsonschema.SchemaError as err:
+            raise ValueError(err.message)
+        return json_schema
+
+    def validate(self, data) -> "QuestionnaireResponse":
+        jsonschema.validate(instance=data, schema=self.json_schema)
+        return QuestionnaireResponse(
+            questionnaire_name=self.name, questionnaire_version=self.version, data=data
+        )
+
+
+class QuestionnaireResponse(BaseModel):
+    id: UUID = Field(default_factory=uuid4)
+    questionnaire_name: str
+    questionnaire_version: str
+    data: dict
+    created_on: datetime = Field(default_factory=now)

src/layers/domain/repository/questionnaire_repository/__init__.py

@@ -1,53 +1 @@
-from pathlib import Path
-
-from domain.core.questionnaire.v2 import Questionnaire
-from domain.repository.errors import ItemNotFound
-from domain.repository.questionnaire_repository.deserialisers import (
-    QUESTION_DESERIALISERS,
-)
-from event.json import json_load
-
-PATH_TO_QUESTIONNAIRES = Path(__file__).parent / "questionnaires"
-
-
-def deserialise_question(question: dict) -> dict:
-    for field, deserialiser in QUESTION_DESERIALISERS.items():
-        value = question.get(field)
-        if value:
-            question[field] = deserialiser(value)
-    return question
-
-
-def version_from_file_path(file_path: Path) -> int:
-    return int(file_path.stem.lstrip("v"))
-
-
-def get_latest_questions_by_name(name: str) -> Path | None:
-    possible_paths = PATH_TO_QUESTIONNAIRES.glob(f"{name}/v*.json")
-    paths_sorted_by_version = sorted(possible_paths, key=version_from_file_path)
-    try:
-        path = paths_sorted_by_version[-1]
-    except IndexError:
-        path = None
-    return path
-
-
-def read_questions(path: Path):
-    with open(path, "r") as fp:
-        raw_questions = json_load(fp)
-    return list(map(deserialise_question, raw_questions))
-
-
-class QuestionnaireRepository:
-
-    def read(self, name: str) -> Questionnaire:
-        path = get_latest_questions_by_name(name=name.lower())
-        if not path:
-            raise ItemNotFound(name, item_type=Questionnaire)
-
-        version = version_from_file_path(path)
-        questions = read_questions(path=path)
-        questionnaire = Questionnaire(name=name, version=version)
-        for question in questions:
-            questionnaire.add_question(**question)
-        return questionnaire
+from .v1 import *  # noqa

src/layers/domain/repository/questionnaire_repository/v1/__init__.py

@@ -0,0 +1,52 @@
+from pathlib import Path
+
+from domain.core.questionnaire.v2 import Questionnaire
+from domain.repository.errors import ItemNotFound
+from event.json import json_load
+
+from .deserialisers import QUESTION_DESERIALISERS
+
+PATH_TO_QUESTIONNAIRES = Path(__file__).parent / "questionnaires"
+
+
+def deserialise_question(question: dict) -> dict:
+    for field, deserialiser in QUESTION_DESERIALISERS.items():
+        value = question.get(field)
+        if value:
+            question[field] = deserialiser(value)
+    return question
+
+
+def version_from_file_path(file_path: Path) -> int:
+    return int(file_path.stem.lstrip("v"))
+
+
+def get_latest_questions_by_name(name: str) -> Path | None:
+    possible_paths = PATH_TO_QUESTIONNAIRES.glob(f"{name}/v*.json")
+    paths_sorted_by_version = sorted(possible_paths, key=version_from_file_path)
+    try:
+        path = paths_sorted_by_version[-1]
+    except IndexError:
+        path = None
+    return path
+
+
+def read_questions(path: Path):
+    with open(path, "r") as fp:
+        raw_questions = json_load(fp)
+    return list(map(deserialise_question, raw_questions))
+
+
+class QuestionnaireRepository:
+
+    def read(self, name: str) -> Questionnaire:
+        path = get_latest_questions_by_name(name=name)
+        if not path:
+            raise ItemNotFound(name, item_type=Questionnaire)
+
+        version = version_from_file_path(path)
+        questions = read_questions(path=path)
+        questionnaire = Questionnaire(name=name, version=version)
+        for question in questions:
+            questionnaire.add_question(**question)
+        return questionnaire

…uestionnaire_repository/deserialisers.py …tionnaire_repository/v1/deserialisers.pysrc/layers/domain/repository/questionnaire_repository/deserialisers.py renamed to src/layers/domain/repository/questionnaire_repository/v1/deserialisers.py src/layers/domain/repository/questionnaire_repository/deserialisers.py renamed to src/layers/domain/repository/questionnaire_repository/v1/deserialisers.py

@@ -1,7 +1,7 @@
 import pytest
 from domain.core.questionnaire.v2 import Questionnaire
-from domain.repository.questionnaire_repository import QuestionnaireRepository
-from domain.repository.questionnaire_repository.questionnaires import (
+from domain.repository.questionnaire_repository.v1 import QuestionnaireRepository
+from domain.repository.questionnaire_repository.v1.questionnaires import (
     QuestionnaireInstance,
 )
 from event.json import json_load