Allow limits on user settings, with per-group overrides

packages/models-library/src/models_library/api_schemas_webserver/users_preferences.py

@@ -1,17 +1,33 @@
-from typing import Any, TypeAlias
+from typing import Annotated, Any
 
 from pydantic import BaseModel, Field
 
 from ..user_preferences import PreferenceIdentifier
 from ._base import InputSchema, OutputSchema
 
 
+class PreferenceConstraints(OutputSchema):
+    """Limits applying to a preference value, used by the frontend to render its widget."""
+
+    ge: int | float | None = None
+    gt: int | float | None = None
+    le: int | float | None = None
+    lt: int | float | None = None
+    max_length: int | None = None
+    min_length: int | None = None
+    multiple_of: int | float | None = None
+    pattern: str | None = None
+
+
 class Preference(OutputSchema):
-    default_value: Any = Field(default=..., description="used by the frontend")
-    value: Any = Field(default=..., description="preference value")
+    default_value: Annotated[Any, Field(description="used by the frontend")]
+    value: Annotated[Any, Field(description="preference value")]
+    constraints: Annotated[PreferenceConstraints | None, Field(description="null when the value is unconstrained")] = (
+        None
+    )
 
 
-AggregatedPreferences: TypeAlias = dict[PreferenceIdentifier, Preference]
+type AggregatedPreferences = dict[PreferenceIdentifier, Preference]
 
 
 class PatchRequestBody(InputSchema):

packages/models-library/src/models_library/user_preferences.py

@@ -1,8 +1,9 @@
+import json
 from enum import auto
-from typing import Annotated, Any, ClassVar, Literal, TypeAlias
+from typing import Annotated, Any, ClassVar, Final, Literal, Self
 
 from common_library.pydantic_fields_extension import get_type
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, create_model
 from pydantic._internal._model_construction import ModelMetaclass
 from pydantic.fields import FieldInfo
 
@@ -29,8 +30,8 @@ def __new__(cls, name, bases, attrs, *args, **kwargs):
         return new_class
 
 
-PreferenceName: TypeAlias = str
-PreferenceIdentifier: TypeAlias = str
+type PreferenceName = str
+type PreferenceIdentifier = str
 
 
 class _ExtendedBaseModel(BaseModel, metaclass=_AutoRegisterMeta): ...
@@ -47,16 +48,79 @@ def __init__(self, preference_name) -> None:
         super().__init__(f"No preference class found for provided {preference_name=}")
 
 
+_ALLOWED_VALUE_CONSTRAINTS: Final[frozenset[str]] = frozenset(
+    {"ge", "gt", "le", "lt", "max_length", "min_length", "multiple_of", "pattern"}
+)
+
+_VALUE_VALIDATOR_CLASSES: Final[dict[tuple[type, str], type[BaseModel]]] = {}
+
+
+class InvalidValueConstraintsError(ValueError):
+    def __init__(self, preference_name: PreferenceName, reason: str) -> None:
+        self.preference_name = preference_name
+        self.reason = reason
+        super().__init__(f"Invalid value constraints for {preference_name=}: {reason}")
+
+
+def _raise_if_not_allowed(preference_name: PreferenceName, constraints: dict[str, Any]) -> None:
+    if rejected := set(constraints) - _ALLOWED_VALUE_CONSTRAINTS:
+        raise InvalidValueConstraintsError(
+            preference_name,
+            f"unsupported {sorted(rejected)}, allowed are {sorted(_ALLOWED_VALUE_CONSTRAINTS)}",
+        )
+
+
 class _BaseUserPreferenceModel(_ExtendedBaseModel):
     preference_type: PreferenceType = Field(..., description="distinguish between the types of preferences")
 
     value: Any = Field(..., description="value of the preference")
 
+    # NOTE: enforced only when setting the value, never by this model itself, so that
+    # values allowed by a looser deployment configuration remain readable.
+    value_constraints: ClassVar[dict[str, Any]] = {}
+
     @classmethod
-    def get_preference_class_from_name(cls, preference_name: PreferenceName) -> type["_BaseUserPreferenceModel"]:
-        preference_class: type[_BaseUserPreferenceModel] | None = cls.registered_user_preference_classes.get(
-            preference_name, None
-        )
+    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
+        super().__pydantic_init_subclass__(**kwargs)
+        _raise_if_not_allowed(cls.get_preference_name(), cls.value_constraints)
+
+    @classmethod
+    def get_value_constraints(cls, overrides: dict[str, Any] | None = None) -> dict[str, Any]:
+        """Constraints declared by the class, with `overrides` taking precedence per key."""
+        constraints = {**cls.value_constraints, **(overrides or {})}
+        _raise_if_not_allowed(cls.get_preference_name(), constraints)
+        return constraints
+
+    @classmethod
+    def build_value_validator(cls, overrides: dict[str, Any] | None = None) -> type[BaseModel]:
+        """Model whose only field validates a preference value against `value_constraints` merged with `overrides`."""
+        # pylint: disable=unsubscriptable-object
+        preference_name = cls.get_preference_name()
+        constraints = cls.get_value_constraints(overrides)
+
+        cache_key = (cls, json.dumps(constraints, sort_keys=True, default=str))
+        if cache_key not in _VALUE_VALIDATOR_CLASSES:
+            value_annotation = cls.model_fields["value"].annotation
+            _VALUE_VALIDATOR_CLASSES[cache_key] = create_model(
+                f"{preference_name}ValueValidator",
+                __base__=BaseModel,
+                value=(Annotated[value_annotation, Field(**constraints)], ...),
+            )
+        return _VALUE_VALIDATOR_CLASSES[cache_key]
+
+    @classmethod
+    def validate_value(cls, value: Any, overrides: dict[str, Any] | None = None) -> None:
+        validator_class = cls.build_value_validator(overrides)
+        try:
+            validator_class(value=value)
+        except TypeError as e:
+            # pydantic reports a constraint that cannot apply to the field type only on use
+            raise InvalidValueConstraintsError(cls.get_preference_name(), f"{e}") from e
+
+    @classmethod
+    def get_preference_class_from_name(cls, preference_name: PreferenceName) -> type[Self]:
+        # NOTE: the registry is untyped (`dict[str, type]`), the annotation below narrows it
+        preference_class: type[Self] | None = cls.registered_user_preference_classes.get(preference_name, None)
         if preference_class is None:
             raise NoPreferenceFoundError(preference_name)
         return preference_class
@@ -117,7 +181,7 @@ def to_db(self) -> dict:
         return self.model_dump(exclude={"preference_type"})
 
 
-AnyUserPreference: TypeAlias = Annotated[
+type AnyUserPreference = Annotated[
     FrontendUserPreference | UserServiceUserPreference,
     Field(discriminator="preference_type"),
 ]

packages/models-library/tests/test_user_preferences.py

@@ -3,19 +3,22 @@
 
 from collections.abc import Iterator
 from pathlib import Path
-from typing import Any
+from typing import Any, ClassVar, Final, NamedTuple
 
 import pytest
+from models_library.api_schemas_webserver.users_preferences import PreferenceConstraints
 from models_library.services import ServiceKey, ServiceVersion
 from models_library.user_preferences import (
+    _ALLOWED_VALUE_CONSTRAINTS,
     FrontendUserPreference,
+    InvalidValueConstraintsError,
     NoPreferenceFoundError,
     PreferenceType,
     UserServiceUserPreference,
     _AutoRegisterMeta,
     _BaseUserPreferenceModel,
 )
-from pydantic import TypeAdapter
+from pydantic import TypeAdapter, ValidationError, create_model
 
 _SERVICE_KEY_AND_VERSION_SAMPLES: list[tuple[ServiceKey, ServiceVersion]] = [
     (
@@ -81,13 +84,16 @@ def test_user_service_preferences(value: Any, mock_file_path: Path):
 
 
 @pytest.fixture
-def unregister_defined_classes() -> Iterator[None]:
-    yield
+def restore_preference_classes_registry() -> Iterator[None]:
     # pylint: disable=protected-access
-    _AutoRegisterMeta.registered_user_preference_classes.pop("Pref1", None)
+    registry = _AutoRegisterMeta.registered_user_preference_classes
+    snapshot = dict(registry)
+    yield
+    registry.clear()
+    registry.update(snapshot)
 
 
-def test__frontend__user_preference(value: Any, unregister_defined_classes: None):
+def test__frontend__user_preference(value: Any, restore_preference_classes_registry: None):
     pref1 = FrontendUserPreference.model_validate({"preference_identifier": "pref_id", "value": value})
     assert isinstance(pref1, FrontendUserPreference)
 
@@ -98,7 +104,7 @@ def test__user_service__user_preference(
     service_key: ServiceKey,
     service_version: ServiceVersion,
     mock_file_path: Path,
-    unregister_defined_classes: None,
+    restore_preference_classes_registry: None,
 ):
     pref1 = UserServiceUserPreference.model_validate(
         {
@@ -116,7 +122,7 @@ def test__user_service__user_preference(
     assert new_instance == pref1
 
 
-def test_redefine_class_with_same_name_is_not_allowed(unregister_defined_classes: None):
+def test_redefine_class_with_same_name_is_not_allowed(restore_preference_classes_registry: None):
     # pylint: disable=unused-variable
     def def_class_1():
         class APreference(_BaseUserPreferenceModel): ...
@@ -132,3 +138,157 @@ class APreference(_BaseUserPreferenceModel): ...
 def test_get_preference_class_from_name_not_found():
     with pytest.raises(NoPreferenceFoundError, match="No preference class found"):
         _BaseUserPreferenceModel.get_preference_class_from_name("__missing_preference_name__")
+
+
+@pytest.fixture
+def capped_preference_class(restore_preference_classes_registry: None) -> type[FrontendUserPreference]:
+    class CappedPreference(FrontendUserPreference):
+        preference_identifier: str = "capped"
+        value: int = 1800
+        value_constraints: ClassVar[dict[str, Any]] = {"le": 10800}
+
+    return CappedPreference
+
+
+@pytest.mark.parametrize(
+    "value, overrides, is_valid",
+    [
+        pytest.param(1800, None, True, id="within_class_constraint"),
+        pytest.param(10800, None, True, id="at_class_constraint"),
+        pytest.param(14400, None, False, id="above_class_constraint"),
+        pytest.param(21600, {"le": 21600}, True, id="override_relaxes_class_constraint"),
+        pytest.param(25200, {"le": 21600}, False, id="above_relaxed_override"),
+        pytest.param(10800, {"le": 7200}, False, id="override_tightens_class_constraint"),
+        pytest.param(30, {"ge": 60}, False, id="override_adds_constraint"),
+        pytest.param("not-an-int", None, False, id="wrong_type"),
+    ],
+)
+def test_validate_value_with_constraint_overrides(
+    capped_preference_class: type[FrontendUserPreference],
+    value: Any,
+    overrides: dict[str, Any] | None,
+    is_valid: bool,
+):
+    if is_valid:
+        capped_preference_class.validate_value(value, overrides)
+    else:
+        with pytest.raises(ValidationError):
+            capped_preference_class.validate_value(value, overrides)
+
+
+def test_validate_value_leaves_preference_class_untouched(
+    capped_preference_class: type[FrontendUserPreference],
+):
+    capped_preference_class.validate_value(21600, {"le": 21600})
+
+    assert capped_preference_class.model_fields["value"].metadata == []
+    assert capped_preference_class.get_default_value() == 1800
+    # a value the deployment allowed must remain readable
+    assert capped_preference_class.model_validate({"value": 21600}).value == 21600
+
+
+def test_build_value_validator_is_cached(
+    capped_preference_class: type[FrontendUserPreference],
+):
+    assert capped_preference_class.build_value_validator(
+        {"le": 21600}
+    ) is capped_preference_class.build_value_validator({"le": 21600})
+    assert capped_preference_class.build_value_validator({"le": 21600}) is not (
+        capped_preference_class.build_value_validator({"le": 7200})
+    )
+
+
+def test_unsupported_constraint_is_rejected(
+    capped_preference_class: type[FrontendUserPreference],
+):
+    with pytest.raises(InvalidValueConstraintsError, match="unsupported"):
+        capped_preference_class.validate_value(1800, {"allow_inf_nan": True})
+
+
+def test_constraint_not_applicable_to_field_type_is_rejected(restore_preference_classes_registry: None):
+    class Pref1(FrontendUserPreference):
+        preference_identifier: str = "pref1"
+        value: str = "a-value"
+
+    with pytest.raises(InvalidValueConstraintsError, match="Unable to apply constraint"):
+        Pref1.validate_value("a-value", {"ge": 1})
+
+
+def test_class_constraints_are_validated_at_class_creation(restore_preference_classes_registry: None):
+    with pytest.raises(InvalidValueConstraintsError, match="unsupported"):
+
+        class Pref1(FrontendUserPreference):  # pylint: disable=unused-variable
+            preference_identifier: str = "pref1"
+            value: int = 1
+            value_constraints: ClassVar[dict[str, Any]] = {"not_a_constraint": 1}
+
+
+def test_nullable_value_accepts_none_with_constraints(restore_preference_classes_registry: None):
+    class Pref1(FrontendUserPreference):
+        preference_identifier: str = "pref1"
+        value: int | None = None
+
+    Pref1.validate_value(None, {"ge": 1})
+    Pref1.validate_value(5, {"ge": 1})
+    with pytest.raises(ValidationError):
+        Pref1.validate_value(0, {"ge": 1})
+
+
+class _ConstraintExample(NamedTuple):
+    name: str
+    value_type: type
+    stored_in_db: dict[str, Any]
+    accepts: Any
+    rejects: Any
+
+
+_CONSTRAINT_EXAMPLES: Final[list[_ConstraintExample]] = [
+    _ConstraintExample("ge", int, {"ge": 60}, 60, 59),
+    _ConstraintExample("gt", int, {"gt": 60}, 61, 60),
+    _ConstraintExample("le", int, {"le": 10800}, 10800, 10801),
+    _ConstraintExample("lt", int, {"lt": 10800}, 10799, 10800),
+    _ConstraintExample("max_length", str, {"max_length": 5}, "abcde", "abcdef"),
+    _ConstraintExample("min_length", str, {"min_length": 3}, "abc", "ab"),
+    _ConstraintExample("multiple_of", int, {"multiple_of": 60}, 120, 121),
+    _ConstraintExample("pattern", str, {"pattern": "^(dark|light)$"}, "dark", "blue"),
+]
+
+
+def test_constraint_examples_cover_all_allowed_constraints():
+    assert {example.name for example in _CONSTRAINT_EXAMPLES} == set(_ALLOWED_VALUE_CONSTRAINTS)
+
+
+def test_frontend_schema_exposes_all_allowed_constraints():
+    assert set(PreferenceConstraints.model_fields) == set(_ALLOWED_VALUE_CONSTRAINTS)
+
+
+@pytest.mark.parametrize(
+    "example",
+    [pytest.param(example, id=example.name) for example in _CONSTRAINT_EXAMPLES],
+)
+def test_every_allowed_constraint_is_enforced(
+    restore_preference_classes_registry: None,
+    example: _ConstraintExample,
+):
+    preference_class: type[FrontendUserPreference] = create_model(
+        "ConstrainedPreference",
+        __base__=FrontendUserPreference,
+        preference_identifier=(str, "constrained"),
+        value=(example.value_type, ...),
+    )
+
+    preference_class.validate_value(example.accepts, example.stored_in_db)
+    with pytest.raises(ValidationError):
+        preference_class.validate_value(example.rejects, example.stored_in_db)
+
+
+def test_get_value_constraints_merges_overrides_per_key(restore_preference_classes_registry: None):
+    class Pref1(FrontendUserPreference):
+        preference_identifier: str = "pref1"
+        value: int = 1800
+        value_constraints: ClassVar[dict[str, Any]] = {"ge": 60, "le": 10800}
+
+    assert Pref1.get_value_constraints() == {"ge": 60, "le": 10800}
+    assert Pref1.get_value_constraints({"le": 21600}) == {"ge": 60, "le": 21600}
+    with pytest.raises(InvalidValueConstraintsError, match="unsupported"):
+        Pref1.get_value_constraints({"not_a_constraint": 1})