refactor: apply suggested changes

Author: janbritz

questionpy_common/constants.py

@@ -2,9 +2,9 @@
 #  QuestionPy is free software released under terms of the MIT license. See LICENSE.md.
 #  (c) Technische Universität Berlin, innoCampus <info@isis.tu-berlin.de>
 import re
-from typing import Annotated, Final
+from typing import Final
 
-from pydantic import ByteSize, Field
+from pydantic import ByteSize
 
 # General.
 KiB: Final[int] = 1024
@@ -28,4 +28,3 @@
 )
 
 ENVIRONMENT_VARIABLE_REGEX: Final[str] = r"[a-zA-Z_][a-zA-Z0-9_]*"
-ENVIRONMENT_VARIABLE = Annotated[str, Field(pattern=f"^{ENVIRONMENT_VARIABLE_REGEX}$")]

questionpy_common/manifest.py

@@ -10,7 +10,7 @@
 from pydantic import BaseModel, ByteSize, PositiveInt, StringConstraints, conset, field_validator
 from pydantic.fields import Field
 
-from questionpy_common.constants import ENVIRONMENT_VARIABLE
+from questionpy_common.constants import ENVIRONMENT_VARIABLE_REGEX
 
 
 class PackageType(StrEnum):
@@ -88,6 +88,9 @@ class PartialPackagePermissions(BaseModel):
     lms_attributes: set[str] | None = None
 
 
+type EnvironmentVariableName = Annotated[str, Field(pattern=f"^{ENVIRONMENT_VARIABLE_REGEX}$")]
+
+
 class SourceManifest(BaseModel):
     """Represents the fields in a package source directory.
 
@@ -113,7 +116,7 @@ class SourceManifest(BaseModel):
     type: PackageType = DEFAULT_PACKAGETYPE
     license: str | None = None
     permissions: PartialPackagePermissions | None = None
-    environment_variables: set[ENVIRONMENT_VARIABLE] | None = None
+    environment_variables: set[EnvironmentVariableName] | None = None
     tags: set[str] = set()
     requirements: str | list[str] | None = None
 

questionpy_server/settings.py

@@ -32,8 +32,8 @@
     SettingsConfigDict,
 )
 
-from questionpy_common.constants import ENVIRONMENT_VARIABLE, ENVIRONMENT_VARIABLE_REGEX, MAX_PACKAGE_SIZE, GiB, MiB
-from questionpy_common.manifest import PartialPackagePermissions, ensure_is_valid_name
+from questionpy_common.constants import ENVIRONMENT_VARIABLE_REGEX, MAX_PACKAGE_SIZE, GiB, MiB
+from questionpy_common.manifest import EnvironmentVariableName, PartialPackagePermissions, ensure_is_valid_name
 from questionpy_server.worker import Worker
 from questionpy_server.worker.impl.subprocess import SubprocessWorker
 
@@ -210,7 +210,7 @@ class PackagePermissionsSettings(BaseModel):
     packages: list[SpecificPackagePermissions] = []
 
 
-class EnvironmentVariables(RootModel[dict[ENVIRONMENT_VARIABLE, str]]):
+class EnvironmentVariables(RootModel[dict[EnvironmentVariableName, str]]):
     interpolation_pattern: ClassVar[re.Pattern] = re.compile(rf"^\$\{{({ENVIRONMENT_VARIABLE_REGEX})}}$")
     """Matches environment variable interpolation syntax for replacement.
 
@@ -219,7 +219,6 @@ class EnvironmentVariables(RootModel[dict[ENVIRONMENT_VARIABLE, str]]):
 
     Example:
         "${MY_VAR}" matches and becomes the value of the environment variable "MY_VAR"
-
     """
 
     escaped_interpolation_pattern: ClassVar[re.Pattern] = re.compile(rf"^\$(\$+\{{{ENVIRONMENT_VARIABLE_REGEX}}})$")

questionpy_server/worker/selector/__init__.py

@@ -1,10 +1,8 @@
 #  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 <info@isis.tu-berlin.de>
-from abc import ABC, abstractmethod
 from typing import NamedTuple
 
-from questionpy_server.cache import LRUCacheMemory
 from questionpy_server.package import Package
 from questionpy_server.settings import PackageSelector, Selectable
 
@@ -36,32 +34,12 @@ def _is_matching(selector: PackageSelector, query: SelectorQuery) -> bool:
     )
 
 
-class Selector[T: Selectable, V](ABC):
-    """Provides helpful methods for getting package and request specific data."""
+def get_matching[T: Selectable](selectables: list[T], query: SelectorQuery) -> T | None:
+    """Gets the first matching selectable, if any.
 
-    def __init__(self, selectables: list[T]):
-        self._cache: LRUCacheMemory[SelectorQuery, V] = LRUCacheMemory(max_size=128)
-        self._selectables = selectables
-
-    def _get_matching(self, query: SelectorQuery) -> T | None:
-        """Gets the first matching selectable, if any.
-
-        It assumes that the selectables are ordered from least specific to most specific.
-        """
-        for selectable in reversed(self._selectables):
-            if _is_matching(selectable.package_selector, query):
-                return selectable
-        return None
-
-    def get(self, query: SelectorQuery) -> V:
-        """Gets the result of the query, which may be cached."""
-        if cached := self._cache.get(query):
-            return cached
-
-        result = self._get(query)
-        self._cache.put(query, result)
-
-        return result
-
-    @abstractmethod
-    def _get(self, query: SelectorQuery) -> V: ...
+    It assumes that the selectables are ordered from least specific to most specific.
+    """
+    for selectable in reversed(selectables):
+        if _is_matching(selectable.package_selector, query):
+            return selectable
+    return None

questionpy_server/worker/selector/environment_variables.py

@@ -2,26 +2,30 @@
 #  The QuestionPy Server is free software released under terms of the MIT license. See LICENSE.md.
 #  (c) Technische Universität Berlin, innoCampus <info@isis.tu-berlin.de>
 from questionpy_common.error import QPyBaseError
-from questionpy_server.settings import EnvironmentVariablesSettings, SpecificPackageEnvironmentVariables
-from questionpy_server.worker.selector import Selector, SelectorQuery
+from questionpy_server.cache import LRUCacheMemory
+from questionpy_server.settings import EnvironmentVariablesSettings
+from questionpy_server.worker.selector import SelectorQuery, get_matching
 
 
 class PackageEnvironmentVariablesError(QPyBaseError):
     pass
 
 
-class EnvironmentVariablesHandler(Selector[SpecificPackageEnvironmentVariables, dict[str, str]]):
+class EnvironmentVariablesHandler:
     """Handles environment variables for a request."""
 
     def __init__(self, settings: EnvironmentVariablesSettings):
-        super().__init__(settings.packages)
-
+        self._cache: LRUCacheMemory[SelectorQuery, dict[str, str]] = LRUCacheMemory(max_size=128)
+        self._environment_variables = settings.packages
         self._global_environment_variables = settings.global_.root
 
-    def _get(self, query: SelectorQuery) -> dict[str, str]:
+    def get(self, query: SelectorQuery) -> dict[str, str]:
+        if cached_environment_variables := self._cache.get(query):
+            return cached_environment_variables
+
         environment_variables = self._global_environment_variables.copy()
 
-        if (specific := self._get_matching(query)) and specific.environment_variables:
+        if (specific := get_matching(self._environment_variables, query)) and specific.environment_variables:
             environment_variables.update(specific.environment_variables.root)
 
         requested_environment_variables = query.package.manifest.environment_variables
@@ -34,4 +38,5 @@ def _get(self, query: SelectorQuery) -> dict[str, str]:
             )
             raise PackageEnvironmentVariablesError(msg)
 
+        self._cache.put(query, environment_variables)
         return environment_variables

questionpy_server/worker/selector/permissions.py

@@ -5,14 +5,15 @@
 
 from questionpy_common.environment import PackagePermissions as EnvironmentPackagePermissions
 from questionpy_common.error import QPyBaseError
+from questionpy_server.cache import LRUCacheMemory
 from questionpy_server.package import Package
 from questionpy_server.settings import (
     CompletePackagePermissions,
     MainProcessExecutionModeValues,
     PackagePermissionsSettings,
     SpecificPackagePermissions,
 )
-from questionpy_server.worker.selector import Selector, SelectorQuery
+from questionpy_server.worker.selector import SelectorQuery, get_matching
 
 _log = logging.getLogger(__name__)
 
@@ -31,12 +32,12 @@ def _has_enough_permissions(allowed: CompletePackagePermissions, requested: Comp
     )
 
 
-class PackagePermissionsHandler(Selector[SpecificPackagePermissions, EnvironmentPackagePermissions]):
+class PackagePermissionsHandler:
     """Handles package permissions for a request."""
 
     def __init__(self, settings: PackagePermissionsSettings):
-        super().__init__(settings.packages)
-
+        self._cache: LRUCacheMemory[SelectorQuery, EnvironmentPackagePermissions] = LRUCacheMemory(max_size=128)
+        self._package_permissions = settings.packages
         self._default_permissions = CompletePackagePermissions()
         self._auto_grant_permissions = settings.auto_grant_permissions
 
@@ -69,11 +70,14 @@ def _get_actual_auto_grant_permissions(self, permissions: SpecificPackagePermiss
         specific_auto_grant_permissions = permissions.auto_grant_permissions.model_dump(exclude_none=True)
         return self._auto_grant_permissions.model_copy(update=specific_auto_grant_permissions)
 
-    def _get(self, query: SelectorQuery) -> EnvironmentPackagePermissions:
+    def get(self, query: SelectorQuery) -> EnvironmentPackagePermissions:
+        if cached_permissions := self._cache.get(query):
+            return cached_permissions
+
         auto_grant_permissions = self._auto_grant_permissions
         requested_permissions = self._get_requested_permissions(query.package)
 
-        if specific_permissions := self._get_matching(query):
+        if specific_permissions := get_matching(self._package_permissions, query):
             auto_grant_permissions = self._get_actual_auto_grant_permissions(specific_permissions)
 
             if specific_permissions.override_permissions:
@@ -89,4 +93,7 @@ def _get(self, query: SelectorQuery) -> EnvironmentPackagePermissions:
         # Only keep explicitly allowed lms attributes.
         requested_permissions.lms_attributes.intersection_update(auto_grant_permissions.lms_attributes)
 
-        return EnvironmentPackagePermissions(**requested_permissions.model_dump())
+        environment_package_permissions = EnvironmentPackagePermissions(**requested_permissions.model_dump())
+        self._cache.put(query, environment_package_permissions)
+
+        return environment_package_permissions