refactor: ensure_package_and_question_state_exist

Author: MHajoha

questionpy_server/__init__.py

@@ -1,9 +1,9 @@
-__version__ = "0.1.0"
-
 #  This file is part of the QuestionPy Server. (https://questionpy.org)
 #  The QuestionPy Server is free software released under terms of the MIT license. See LICENSE.md.
 #  (c) Technische Universität Berlin, innoCampus <[email protected]>
 
 from questionpy_server.worker.pool import WorkerPool
 
-__all__ = ["WorkerPool"]
+__version__ = "0.1.0"
+
+__all__ = ["WorkerPool", "__version__"]

questionpy_server/api/routes/_attempts.py

@@ -8,25 +8,25 @@
 
 from questionpy_common.environment import RequestUser
 from questionpy_server.api.models import AttemptScoreArguments, AttemptStartArguments, AttemptViewArguments
-from questionpy_server.decorators import ensure_package_and_question_state_exist
+from questionpy_server.app import QPyServer
+from questionpy_server.decorators import ensure_required_parts
 from questionpy_server.package import Package
 from questionpy_server.web import json_response
 from questionpy_server.worker.runtime.package_location import ZipPackageLocation
 
 if TYPE_CHECKING:
-    from questionpy_server.app import QPyServer
     from questionpy_server.worker.worker import Worker
 
 
 attempt_routes = web.RouteTableDef()
 
 
 @attempt_routes.post(r"/packages/{package_hash:\w+}/attempt/start")  # type: ignore[arg-type]
-@ensure_package_and_question_state_exist
+@ensure_required_parts
 async def post_attempt_start(
     request: web.Request, package: Package, question_state: bytes, data: AttemptStartArguments
 ) -> web.Response:
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
 
     package_path = await package.get_path()
     worker: Worker
@@ -37,11 +37,11 @@ async def post_attempt_start(
 
 
 @attempt_routes.post(r"/packages/{package_hash:\w+}/attempt/view")  # type: ignore[arg-type]
-@ensure_package_and_question_state_exist
+@ensure_required_parts
 async def post_attempt_view(
     request: web.Request, package: Package, question_state: bytes, data: AttemptViewArguments
 ) -> web.Response:
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
 
     package_path = await package.get_path()
     worker: Worker
@@ -58,11 +58,11 @@ async def post_attempt_view(
 
 
 @attempt_routes.post(r"/packages/{package_hash:\w+}/attempt/score")  # type: ignore[arg-type]
-@ensure_package_and_question_state_exist
+@ensure_required_parts
 async def post_attempt_score(
     request: web.Request, package: Package, question_state: bytes, data: AttemptScoreArguments
 ) -> web.Response:
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
 
     package_path = await package.get_path()
     worker: Worker

questionpy_server/api/routes/_files.py

@@ -6,35 +6,35 @@
 from aiohttp import web
 from aiohttp.web_exceptions import HTTPNotImplemented
 
-from questionpy_server.decorators import ensure_package_and_question_state_exist
+from questionpy_server.app import QPyServer
+from questionpy_server.decorators import ensure_package
 from questionpy_server.package import Package
 from questionpy_server.worker.runtime.package_location import ZipPackageLocation
 
 if TYPE_CHECKING:
-    from questionpy_server.app import QPyServer
     from questionpy_server.worker.worker import Worker
 
 file_routes = web.RouteTableDef()
 
 
 @file_routes.post(r"/packages/{package_hash}/file/{namespace}/{short_name}/{path:static/.*}")  # type: ignore[arg-type]
-@ensure_package_and_question_state_exist
+@ensure_package
 async def post_attempt_start(request: web.Request, package: Package) -> web.Response:
-    qpy_server: QPyServer = request.app["qpy_server_app"]
+    qpy_server = request.app[QPyServer.APP_KEY]
     namespace = request.match_info["namespace"]
     short_name = request.match_info["short_name"]
     path = request.match_info["path"]
 
     if package.manifest.namespace != namespace or package.manifest.short_name != short_name:
         # TODO: Support static files in non-main packages by using namespace and short_name.
-        raise HTTPNotImplemented(reason="Static file retrieval from non-main packages is not supported yet.")
+        raise HTTPNotImplemented(text="Static file retrieval from non-main packages is not supported yet.")
 
     worker: Worker
     async with qpy_server.worker_pool.get_worker(ZipPackageLocation(await package.get_path()), 0, None) as worker:
         try:
             file = await worker.get_static_file(path)
         except FileNotFoundError as e:
-            raise web.HTTPNotFound(reason="File not found.") from e
+            raise web.HTTPNotFound(text="File not found.") from e
 
     return web.Response(
         body=file.data,

questionpy_server/api/routes/_packages.py

@@ -9,21 +9,21 @@
 
 from questionpy_common.environment import RequestUser
 from questionpy_server.api.models import QuestionCreateArguments, QuestionEditFormResponse, RequestBaseData
-from questionpy_server.decorators import ensure_package_and_question_state_exist
+from questionpy_server.app import QPyServer
+from questionpy_server.decorators import ensure_package, ensure_required_parts
 from questionpy_server.package import Package
 from questionpy_server.web import json_response
 from questionpy_server.worker.runtime.package_location import ZipPackageLocation
 
 if TYPE_CHECKING:
-    from questionpy_server.app import QPyServer
     from questionpy_server.worker.worker import Worker
 
 package_routes = web.RouteTableDef()
 
 
 @package_routes.get("/packages")
 async def get_packages(request: web.Request) -> web.Response:
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
 
     packages = qpyserver.package_collection.get_packages()
     data = [package.get_info() for package in packages]
@@ -33,22 +33,22 @@ async def get_packages(request: web.Request) -> web.Response:
 
 @package_routes.get(r"/packages/{package_hash:\w+}")
 async def get_package(request: web.Request) -> web.Response:
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
 
-    try:
-        package = qpyserver.package_collection.get(request.match_info["package_hash"])
-        return json_response(data=package.get_info())
-    except FileNotFoundError as error:
-        raise HTTPNotFound from error
+    package = qpyserver.package_collection.get(request.match_info["package_hash"])
+    if not package:
+        raise HTTPNotFound
+
+    return json_response(data=package.get_info())
 
 
 @package_routes.post(r"/packages/{package_hash:\w+}/options")  # type: ignore[arg-type]
-@ensure_package_and_question_state_exist
+@ensure_required_parts
 async def post_options(
-    request: web.Request, package: Package, question_state: bytes | None, data: RequestBaseData
+    request: web.Request, package: Package, data: RequestBaseData, question_state: bytes | None = None
 ) -> web.Response:
     """Get the options form definition that allow a question creator to customize a question."""
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
 
     package_path = await package.get_path()
     worker: Worker
@@ -61,11 +61,11 @@ async def post_options(
 
 
 @package_routes.post(r"/packages/{package_hash:\w+}/question")  # type: ignore[arg-type]
-@ensure_package_and_question_state_exist
+@ensure_required_parts
 async def post_question(
     request: web.Request, data: QuestionCreateArguments, package: Package, question_state: bytes | None = None
 ) -> web.Response:
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
 
     package_path = await package.get_path()
     worker: Worker
@@ -84,7 +84,7 @@ async def post_question_migrate(_request: web.Request) -> web.Response:
 
 
 @package_routes.post(r"/package-extract-info")  # type: ignore[arg-type]
-@ensure_package_and_question_state_exist
+@ensure_package
 async def package_extract_info(_request: web.Request, package: Package) -> web.Response:
     """Get package information."""
     return json_response(data=package.get_info(), status=201)

questionpy_server/api/routes/_status.py

@@ -2,25 +2,20 @@
 #  The QuestionPy Server 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 aiohttp import web
 
 from questionpy_server import __version__
 from questionpy_server.api.models import ServerStatus, Usage
+from questionpy_server.app import QPyServer
 from questionpy_server.web import json_response
 
-if TYPE_CHECKING:
-    from questionpy_server.app import QPyServer
-
-
 status_routes = web.RouteTableDef()
 
 
 @status_routes.get(r"/status")
 async def get_server_status(request: web.Request) -> web.Response:
     """Get server status."""
-    qpyserver: QPyServer = request.app["qpy_server_app"]
+    qpyserver = request.app[QPyServer.APP_KEY]
     status = ServerStatus(
         version=__version__,
         allow_lms_packages=qpyserver.settings.webservice.allow_lms_packages,

questionpy_server/app.py

@@ -3,24 +3,28 @@
 #  (c) Technische Universität Berlin, innoCampus <[email protected]>
 
 from asyncio import create_task
-from typing import Any
+from typing import Any, ClassVar
 
 from aiohttp import web
 
 from . import __version__
-from .api.routes import routes
 from .cache import FileLimitLRU
 from .collector import PackageCollection
 from .settings import Settings
 from .worker.pool import WorkerPool
 
 
-class QPyServer:
+class QPyServer(web.AppKey["QPyServer"]):
+    APP_KEY: ClassVar[web.AppKey["QPyServer"]] = web.AppKey("qpy_server_app")
+
     def __init__(self, settings: Settings):
+        # We import here, so we don't have to work around circular imports.
+        from .api.routes import routes  # noqa: PLC0415
+
         self.settings: Settings = settings
         self.web_app = web.Application(client_max_size=settings.webservice.max_main_size)
         self.web_app.add_routes(routes)
-        self.web_app["qpy_server_app"] = self
+        self.web_app[self.APP_KEY] = self
 
         self.worker_pool = WorkerPool(
             settings.worker.max_workers, settings.worker.max_memory, worker_type=settings.worker.type

questionpy_server/cache.py

@@ -19,9 +19,14 @@ class File(NamedTuple):
     size: int
 
 
-class SizeError(Exception):
-    def __init__(self, message: str = "", max_size: int = 0, actual_size: int = 0):
-        super().__init__(message)
+class CacheItemTooLargeError(Exception):
+    def __init__(self, key: str, actual_size: int, max_size: int):
+        readable_actual = ByteSize(actual_size).human_readable()
+        readable_max = ByteSize(max_size).human_readable()
+        super().__init__(
+            f"Unable to cache item '{key}' with size '{readable_actual}' because it exceeds the maximum "
+            f"allowed size of '{readable_max}'"
+        )
 
         self.max_size = max_size
         self.actual_size = actual_size
@@ -146,12 +151,7 @@ async def put(self, key: str, value: bytes) -> Path:
         if size > self.max_size:
             # If we allowed this, the loop at the end would remove all items from the dictionary,
             # so we raise an error to allow exceptions for this case.
-            msg = f"Item itself exceeds maximum allowed size of {ByteSize(self.max_size).human_readable()}"
-            raise SizeError(
-                msg,
-                max_size=self.max_size,
-                actual_size=size,
-            )
+            raise CacheItemTooLargeError(key, size, self.max_size)
 
         async with self._lock:
             # Save the bytes on filesystem.

questionpy_server/collector/__init__.py

@@ -2,7 +2,7 @@
 #  The QuestionPy Server is free software released under terms of the MIT license. See LICENSE.md.
 #  (c) Technische Universität Berlin, innoCampus <[email protected]>
 
-from questionpy_server.collector.package_collection import PackageCollection
+from questionpy_server.collector._package_collection import PackageCollection
 
 __all__ = [
     "PackageCollection",

…y_server/collector/package_collection.py …_server/collector/_package_collection.pyquestionpy_server/collector/package_collection.py renamed to questionpy_server/collector/_package_collection.py questionpy_server/collector/package_collection.py renamed to questionpy_server/collector/_package_collection.py

@@ -80,20 +80,13 @@ async def put(self, package_container: "HashContainer") -> "Package":
         """
         return await self._lms_collector.put(package_container)
 
-    def get(self, package_hash: str) -> "Package":
+    def get(self, package_hash: str) -> "Package | None":
         """Returns a package if it exists.
 
         Args:
-          package_hash (str): hash value of the package
-
-        Returns:
-          path to the package
+          package_hash: hash value of the package
         """
-        # Check if package was indexed
-        if package := self._indexer.get_by_hash(package_hash):
-            return package
-
-        raise FileNotFoundError
+        return self._indexer.get_by_hash(package_hash)
 
     def get_by_identifier(self, identifier: str) -> dict[SemVer, "Package"]:
         """Returns a dict of packages with the given identifier and available versions.
@@ -106,20 +99,14 @@ def get_by_identifier(self, identifier: str) -> dict[SemVer, "Package"]:
         """
         return self._indexer.get_by_identifier(identifier)
 
-    def get_by_identifier_and_version(self, identifier: str, version: SemVer) -> "Package":
+    def get_by_identifier_and_version(self, identifier: str, version: SemVer) -> "Package | None":
         """Returns a package with the given identifier and version.
 
         Args:
-          identifier (str): identifier of the package
-          version (str): version of the package
-
-        Returns:
-          package
+          identifier: identifier of the package
+          version: version of the package
         """
-        if package := self._indexer.get_by_identifier_and_version(identifier, version):
-            return package
-
-        raise FileNotFoundError
+        return self._indexer.get_by_identifier_and_version(identifier, version)
 
     def get_packages(self) -> set["Package"]:
         """Returns a set of all available packages.