refactor: apply suggested changes

Author: janbritz

questionpy_server/settings.py

@@ -212,7 +212,25 @@ class PackagePermissionsSettings(BaseModel):
 
 class EnvironmentVariables(RootModel[dict[ENVIRONMENT_VARIABLE, str]]):
     interpolation_pattern: ClassVar[re.Pattern] = re.compile(rf"^\$\{{({ENVIRONMENT_VARIABLE_REGEX})}}$")
+    """Matches environment variable interpolation syntax for replacement.
+
+    Identifies strings in the format ${VAR_NAME} that should be replaced with the
+    corresponding environment variable value.
+
+    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}}})$")
+    """Matches escaped interpolation syntax to prevent variable replacement.
+
+    Identifies strings with escaped dollar signs that should be unescaped rather
+    than treated as environment variable references.
+
+    Example:
+        "$${MY_VAR}" matches and becomes "${MY_VAR}" (literal string)
+    """
 
     @model_validator(mode="after")
     def check_environment_variables(self) -> Self:

questionpy_server/worker/__init__.py

@@ -74,7 +74,7 @@ def __init__(self, **kwargs: Unpack[WorkerArgs]) -> None:
         self.package = kwargs["package"]
         self.worker_home = kwargs["worker_home"]
         self.permissions = kwargs["permissions"]
-        self.environment_variables: dict[str, str] = kwargs["environment_variables"]
+        self.environment_variables = kwargs["environment_variables"]
 
         self.state = WorkerState.NOT_RUNNING
         self.loaded_packages: list[LoadedPackage] = []

questionpy_server/worker/selector/__init__.py

@@ -4,6 +4,7 @@
 from abc import ABC, abstractmethod
 from typing import NamedTuple
 
+from questionpy_common.error import QPyBaseError
 from questionpy_server.cache import LRUCacheMemory
 from questionpy_server.package import Package
 from questionpy_server.settings import PackageSelector, Selectable
@@ -37,19 +38,26 @@ 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 __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:
-        if cached_environment_variables := self._cache.get(query):
-            return cached_environment_variables
+        """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)

questionpy_server/worker/selector/environment_variables.py

@@ -11,13 +11,14 @@ class PackageEnvironmentVariablesError(QPyBaseError):
 
 
 class EnvironmentVariablesHandler(Selector[SpecificPackageEnvironmentVariables, dict[str, str]]):
+    """Handles environment variables for a request."""
+
     def __init__(self, settings: EnvironmentVariablesSettings):
         super().__init__(settings.packages)
 
         self._global_environment_variables = settings.global_.root
 
     def _get(self, query: SelectorQuery) -> dict[str, str]:
-        """Gets the environment variables for a package."""
         environment_variables = self._global_environment_variables.copy()
 
         if (specific := self._get_matching(query)) and specific.environment_variables:
@@ -34,11 +35,3 @@ def _get(self, query: SelectorQuery) -> dict[str, str]:
             raise PackageEnvironmentVariablesError(msg)
 
         return environment_variables
-
-    def get(self, query: SelectorQuery) -> dict[str, str]:
-        """Gets the environment variables for a package.
-
-        Raises:
-            PackageEnvironmentVariablesError: If the server is not providing the requested environment variables.
-        """
-        return super().get(query)

questionpy_server/worker/selector/permissions.py

@@ -12,7 +12,7 @@
     PackagePermissionsSettings,
     SpecificPackagePermissions,
 )
-from questionpy_server.worker.selector import Selector, SelectorQuery
+from questionpy_server.worker.selector import Selector, SelectorError, SelectorQuery
 
 _log = logging.getLogger(__name__)
 
@@ -90,11 +90,3 @@ def _get(self, query: SelectorQuery) -> EnvironmentPackagePermissions:
         requested_permissions.lms_attributes.intersection_update(auto_grant_permissions.lms_attributes)
 
         return EnvironmentPackagePermissions(**requested_permissions.model_dump())
-
-    def get(self, query: SelectorQuery) -> EnvironmentPackagePermissions:
-        """Gets the effective permissions for a package.
-
-        Raises:
-            PackagePermissionError: If the requested package permissions are not granted.
-        """
-        return super().get(query)