feat(webserver): call custom javascript

Author: janbritz

questionpy_sdk/webserver/controllers/attempt/controller.py

@@ -3,6 +3,7 @@
 #  (c) Technische Universität Berlin, innoCampus <[email protected]>
 
 import random
+import re
 from enum import StrEnum
 from typing import TYPE_CHECKING, Any, Literal, TypedDict, overload
 
@@ -11,7 +12,7 @@
 from pydantic.dataclasses import dataclass
 
 from questionpy import AttemptModel, AttemptScoredModel, AttemptStartedModel, ScoreModel
-from questionpy_common.api.attempt import ScoringCode
+from questionpy_common.api.attempt import FeedbackType, JsModuleCall, ScoringCode
 from questionpy_sdk.webserver.constants import DEFAULT_REQUEST_USER
 from questionpy_sdk.webserver.controllers.base import BaseController
 from questionpy_sdk.webserver.controllers.errors import MissingAttemptDataError, MissingAttemptStateError
@@ -42,6 +43,9 @@ class AttemptTemplateContext(TypedDict):
     general_feedback: str | None
     specific_feedback: str | None
     right_answer: str | None
+    display_options: QuestionDisplayOptions
+    import_map: dict[str, str]
+    javascript_calls: list[JsModuleCall]
 
 
 @dataclass
@@ -133,14 +137,23 @@ async def _render_ui(
             display_options.right_answer = display_options.correctness = False
 
         # Render UI
-        renderer_args = (attempt.ui.placeholders, display_options, await self._get_attempt_seed(), last_attempt_data)
+        renderer_args = (
+            attempt.ui.placeholders,
+            display_options,
+            self.qpy_url_replacer,
+            await self._get_attempt_seed(),
+            last_attempt_data,
+        )
         html, errors = QuestionFormulationUIRenderer(attempt.ui.formulation, *renderer_args).render()
 
         template_context: AttemptTemplateContext = {
             "formulation": html,
             "general_feedback": None,
             "specific_feedback": None,
             "right_answer": None,
+            "display_options": display_options,
+            "import_map": await self._get_import_map(),
+            "javascript_calls": self._get_js_calls(attempt, display_options),
         }
 
         render_errors: SectionErrorMap = {}
@@ -156,6 +169,36 @@ async def _render_ui(
 
         return template_context, render_errors
 
+    async def _get_import_map(self) -> dict[str, str]:
+        worker: Worker
+        async with self._worker_pool.get_worker(self._package_location, 0, None) as worker:
+            dependencies = worker.get_loaded_packages(only_with_hash=False)
+
+        return {
+            f"@{dependency.namespace}/{dependency.short_name}/":
+                str(self.generate_api_url(
+                    "file",
+                    namespace=dependency.namespace,
+                    short_name=dependency.short_name,
+                    path="static/js/",
+                ))
+            for dependency in dependencies
+        }  # fmt: skip
+
+    def _get_js_calls(self, attempt: AttemptModel, display_options: QuestionDisplayOptions) -> list[JsModuleCall]:
+        feedback_map = {
+            FeedbackType.GENERAL_FEEDBACK: display_options.general_feedback,
+            FeedbackType.SPECIFIC_FEEDBACK: display_options.specific_feedback,
+            FeedbackType.RIGHT_ANSWER: display_options.right_answer,
+        }
+
+        return [
+            call
+            for call in attempt.ui.javascript_calls
+            if (call.if_role is None or call.if_role in display_options.roles)
+            and (call.if_feedback_type is None or feedback_map[call.if_feedback_type])
+        ]
+
     @property
     def _attempt_template(self) -> jinja2.Template:
         loader = jinja2.PackageLoader("questionpy_sdk.webserver")
@@ -225,3 +268,13 @@ async def _get_last_attempt_data(self, *, allow_missing: bool = False) -> dict[s
             if allow_missing:
                 return None
             raise MissingAttemptDataError from err
+
+    def qpy_url_replacer(self, match: re.Match[str]) -> str:
+        return str(
+            self.generate_api_url(
+                "file",
+                namespace=match.group(2),
+                short_name=match.group(3),
+                path=match.group(1),
+            )
+        )

questionpy_sdk/webserver/controllers/attempt/question_ui.py

@@ -6,7 +6,7 @@
 
 import re
 from random import Random
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import lxml.html
 import lxml.html.clean
@@ -29,6 +29,9 @@
     XMLSyntaxError,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
 _XHTML_NAMESPACE: str = "http://www.w3.org/1999/xhtml"
 _QPY_NAMESPACE: str = "http://questionpy.org/ns/question"
 
@@ -208,10 +211,12 @@ def __init__(
         xml: str,
         placeholders: dict[str, str],
         options: QuestionDisplayOptions,
+        qpy_url_replacer: Callable[[re.Match[str]], str],
         seed: int | None = None,
         attempt: dict | None = None,
     ) -> None:
         self._html: str | None = None
+        self._qpy_url_replacer = qpy_url_replacer
 
         xml = self._replace_qpy_urls(xml)
         self._error_collector = _RenderErrorCollector(xml, placeholders)
@@ -265,7 +270,11 @@ def render(self) -> tuple[str, RenderErrorCollection]:
 
     def _replace_qpy_urls(self, xml: str) -> str:
         """Replace QPY-URLs to package files with SDK-URLs."""
-        return re.sub(r"qpy://(static|static-private)/((?:[a-z_][a-z0-9_]{0,126}/){2})", r"/worker/\2file/\1/", xml)
+        return re.sub(
+            r"qpy://((?:static|static-private)/)([a-z_]\w{0,126})/([a-z_]\w{0,126})/",
+            self._qpy_url_replacer,
+            xml,
+        )
 
     def _resolve_placeholders(self) -> None:
         """Replace placeholder PIs such as `<?p my_key plain?>` with the appropriate value from `self.placeholders`.
@@ -492,10 +501,11 @@ def __init__(
         xml: str,
         placeholders: dict[str, str],
         options: QuestionDisplayOptions,
+        qpy_url_replacer: Callable[[re.Match[str]], str],
         seed: int | None = None,
         attempt: dict | None = None,
     ) -> None:
-        super().__init__(xml, placeholders, options, seed, attempt)
+        super().__init__(xml, placeholders, options, qpy_url_replacer, seed, attempt)
         self.metadata = self._get_metadata()
 
     def _get_metadata(self) -> QuestionMetadata:

questionpy_sdk/webserver/controllers/file.py

@@ -0,0 +1,22 @@
+#  This file is part of the QuestionPy SDK. (https://questionpy.org)
+#  The QuestionPy SDK is free software released under terms of the MIT license. See LICENSE.md.
+#  (c) Technische Universität Berlin, innoCampus <[email protected]>
+from typing import TYPE_CHECKING
+
+from questionpy_sdk.webserver.controllers.base import BaseController
+from questionpy_server.worker import PackageFileData
+
+if TYPE_CHECKING:
+    from questionpy_server.worker import Worker
+
+
+class FileController(BaseController):
+    async def get_static_file(self, namespace: str, short_name: str, path: str) -> PackageFileData:
+        if self._manifest.namespace != namespace or self._manifest.short_name != short_name:
+            # TODO: Support static files in non-main packages by using namespace and short_name.
+            msg = "Static file retrieval from non-main packages is not supported yet."
+            raise ValueError(msg)
+
+        worker: Worker
+        async with self._worker_pool.get_worker(self._package_location, 0, None) as worker:
+            return await worker.get_static_file(path)

questionpy_sdk/webserver/routes/__init__.py

@@ -3,7 +3,8 @@
 #  (c) Technische Universität Berlin, innoCampus <[email protected]>
 
 from .attempt import routes as attempt_routes
+from .file import routes as file_routes
 from .manifest import routes as manifest_routes
 from .options import routes as options_routes
 
-api_routes = (attempt_routes, manifest_routes, options_routes)
+api_routes = (attempt_routes, manifest_routes, options_routes, file_routes)

questionpy_sdk/webserver/routes/file.py

@@ -0,0 +1,35 @@
+#  This file is part of the QuestionPy SDK. (https://questionpy.org)
+#  The QuestionPy SDK is free software released under terms of the MIT license. See LICENSE.md.
+#  (c) Technische Universität Berlin, innoCampus <[email protected]>
+
+from aiohttp import web
+from aiohttp.web_exceptions import HTTPNotImplemented
+
+from questionpy_sdk.webserver.controllers.file import FileController
+from questionpy_sdk.webserver.routes.base import BaseView
+
+routes = web.RouteTableDef()
+
+
+@routes.view(r"/file/{namespace}/{short_name}/{path:(static|static-private)/.*}", name="file")
+class FileView(BaseView["FileController"]):
+    controller_class = FileController
+
+    async def get(self) -> web.Response:
+        """Gets the file from the specified package at the given path."""
+        namespace = self.request.match_info["namespace"]
+        short_name = self.request.match_info["short_name"]
+        path = self.request.match_info["path"]
+
+        try:
+            file = await self.controller.get_static_file(namespace, short_name, path)
+        except FileNotFoundError as e:
+            raise web.HTTPNotFound(text="File not found.") from e
+        except ValueError as e:
+            raise HTTPNotImplemented(text=str(e)) from e
+
+        return web.Response(
+            body=file.data,
+            content_type=file.mime_type,
+            headers={"Cache-Control": "no-store"},
+        )