Enabled config validation at all things

Author: eadwinCode

ellar/core/conf/app_settings_models.py

@@ -17,12 +17,38 @@
 from ellar.pydantic import ENCODERS_BY_TYPE as encoders_by_type
 from ellar.pydantic import AllowTypeOfSource, field_validator
 from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.middleware import Middleware
+from starlette.requests import HTTPConnection, Request
 from starlette.websockets import WebSocketClose
 from typing_extensions import Annotated
 
 if t.TYPE_CHECKING:  # pragma: no cover
     from ellar.app.main import App
 
+TEMPLATE_TYPE = t.Dict[
+    str, t.Union[t.Callable[..., t.Any], t.Dict[str, t.Callable[..., t.Any]]]
+]
+
+
+_TemplateValidator = AllowTypeOfSource(
+    validator=lambda s, v: callable(v),
+    error_message=lambda v: f"Expected a callable, got {type(v)}",
+)
+
+_VersioningValidator = AllowTypeOfSource(
+    error_message=lambda s, v: f"Expected BaseAPIVersioning, received: {type(v)}"
+)
+
+_MiddlewareValidator = AllowTypeOfSource(
+    validator=lambda s, v: isinstance(v, (s, Middleware)),
+    error_message=lambda s,
+    v: f"Expected EllarMiddleware or Starlette Middleware object, received: {type(v)}",
+)
+
+_ExceptionHandlerValidator = AllowTypeOfSource(
+    error_message=lambda s, v: f"Expected IExceptionHandler object, received: {type(v)}"
+)
+
 
 async def _not_found(
     scope: TScope, receive: TReceive, send: TSend
@@ -42,7 +68,7 @@ async def _not_found(
     await response(scope, receive, send)
 
 
-class ConfigValidationSchema(Serializer, ConfigDefaultTypesMixin):
+class ConfigSchema(Serializer, ConfigDefaultTypesMixin):
     _filter = SerializerFilter(
         exclude={
             "EXCEPTION_HANDLERS_DECORATOR",
@@ -52,7 +78,12 @@ class ConfigValidationSchema(Serializer, ConfigDefaultTypesMixin):
         }
     )
 
-    model_config = {"validate_assignment": True, "from_attributes": True}
+    model_config = {
+        "validate_assignment": True,
+        "validate_default": True,
+        "from_attributes": True,
+        "extra": "allow",
+    }
 
     OVERRIDE_CORE_SERVICE: t.List[Annotated[ProviderConfig, AllowTypeOfSource()]] = []
 
@@ -73,8 +104,21 @@ class ConfigValidationSchema(Serializer, ConfigDefaultTypesMixin):
     JINJA_TEMPLATES_OPTIONS: t.Dict[str, t.Any] = {}
     JINJA_LOADERS: t.List[JinjaLoaderType] = []
 
+    TEMPLATES_CONTEXT_PROCESSORS: t.List[
+        Annotated[
+            t.Callable[[t.Union[Request, HTTPConnection]], t.Dict[str, t.Any]],
+            _TemplateValidator,
+        ]
+    ] = [  # type:ignore[assignment]
+        "ellar.core.templating.context_processors:request_context",
+        "ellar.core.templating.context_processors:user",
+        "ellar.core.templating.context_processors:request_state",
+    ]
+
     # Application route versioning scheme
-    VERSIONING_SCHEME: t.Optional[IAPIVersioning] = None
+    VERSIONING_SCHEME: t.Optional[Annotated[IAPIVersioning, _VersioningValidator]] = (
+        None
+    )
 
     REDIRECT_SLASHES: bool = False
 
@@ -96,9 +140,23 @@ class ConfigValidationSchema(Serializer, ConfigDefaultTypesMixin):
     ALLOWED_HOSTS: t.List[str] = ["*"]
     REDIRECT_HOST: bool = True
 
-    MIDDLEWARE: t.List[IEllarMiddleware] = []
-
-    EXCEPTION_HANDLERS: t.List[IExceptionHandler] = []
+    MIDDLEWARE: t.List[Annotated[IEllarMiddleware, _MiddlewareValidator]] = [
+        "ellar.core.middleware.trusted_host:trusted_host_middleware",
+        "ellar.core.middleware.cors:cors_middleware",
+        "ellar.core.middleware.errors:server_error_middleware",
+        "ellar.core.middleware.versioning:versioning_middleware",
+        "ellar.auth.middleware.session:session_middleware",
+        "ellar.auth.middleware.auth:identity_middleware",
+        "ellar.core.middleware.exceptions:exception_middleware",
+    ]
+    # A dictionary mapping either integer status codes,
+    # or exception class types onto callables which handle the exceptions.
+    # Exception handler callables should be of the form
+    # `handler(context:IExecutionContext, exc: Exception) -> response`
+    # and may be either standard functions, or async functions.
+    EXCEPTION_HANDLERS: t.List[
+        Annotated[IExceptionHandler, _ExceptionHandlerValidator]
+    ] = ["ellar.core.exceptions:error_404_handler"]
 
     # Default not found handler
     DEFAULT_NOT_FOUND_HANDLER: ASGIApp = _not_found
@@ -118,8 +176,8 @@ class ConfigValidationSchema(Serializer, ConfigDefaultTypesMixin):
     # logging Level
     LOG_LEVEL: t.Optional[log_levels] = log_levels.info
 
-    TEMPLATE_FILTERS: t.Dict[str, t.Callable[..., t.Any]] = {}
-    TEMPLATE_GLOBAL_FILTERS: t.Dict[str, t.Callable[..., t.Any]] = {}
+    TEMPLATE_FILTERS: TEMPLATE_TYPE = {}
+    TEMPLATE_GLOBAL_FILTERS: TEMPLATE_TYPE = {}
 
     LOGGING: t.Optional[t.Dict[str, t.Any]] = None
     CACHES: t.Dict[str, t.Any] = {}

ellar/core/conf/config.py

@@ -1,9 +1,8 @@
 import typing as t
 
-from ellar.common.compatible.dict import AttributeDictAccessMixin
 from ellar.common.constants import ELLAR_CONFIG_MODULE
 from ellar.common.types import VT
-from ellar.core.conf.app_settings_models import ConfigValidationSchema
+from ellar.core.conf.app_settings_models import ConfigSchema
 from ellar.core.conf.mixins import ConfigDefaultTypesMixin
 from ellar.utils.importer import import_from_string
 from starlette.config import environ
@@ -13,8 +12,13 @@ class ConfigRuntimeError(RuntimeError):
     pass
 
 
-class Config(AttributeDictAccessMixin, dict, ConfigDefaultTypesMixin):
-    __slots__ = ("config_module",)
+class Config(ConfigDefaultTypesMixin):
+    __slots__ = (
+        "_config_module",
+        "_schema",
+    )
+
+    _initialized: bool = False
 
     def __init__(
         self,
@@ -23,52 +27,90 @@ def __init__(
         **mapping: t.Any,
     ):
         """
-        Creates a new instance of Configuration object with the given values.
+        Creates a new instance of a Configuration object with the given values.
         """
-        super().__init__()
-        self.config_module = config_module or environ.get(ELLAR_CONFIG_MODULE, None)
 
-        self.clear()
+        self._config_module = config_module or environ.get(ELLAR_CONFIG_MODULE, None)
 
-        self._load_config_module(config_prefix or "")
+        data = self._load_config_module(config_prefix or "")
+        data.update(**mapping)
 
-        self.update(**mapping)
+        self._schema = ConfigSchema.model_validate(data, from_attributes=True)
+        self._initialized = True
 
-        validate_config = ConfigValidationSchema.model_validate(
-            self, from_attributes=True
-        )
-        self.update(validate_config.serialize())
+    @property
+    def config_module(self) -> t.Optional[str]:
+        return self._config_module
 
-    def _load_config_module(self, prefix: str) -> None:
+    def _load_config_module(self, prefix: str) -> dict:
+        data = {}
         _prefix = prefix.upper()
-        if self.config_module:
+
+        if self._config_module:
             try:
-                mod = import_from_string(self.config_module)
+                mod = import_from_string(self._config_module)
                 for setting in dir(mod):
                     if setting.isupper() and setting.startswith(_prefix):
-                        self[setting.replace(_prefix, "")] = getattr(mod, setting)
+                        data[setting.replace(_prefix, "")] = getattr(mod, setting)
             except Exception as ex:
                 raise ConfigRuntimeError(str(ex)) from ex
 
-    def set_defaults(self, **kwargs: t.Any) -> "Config":
-        for k, v in kwargs.items():
-            self.setdefault(k, v)
-        return self
+        return data
 
     def __repr__(self) -> str:  # pragma: no cover
-        hidden_values = {key: "..." for key in self.keys()}
-        return f"<Configuration {repr(hidden_values)}, settings_module: {self.config_module}>"
-
-    def __setattr__(self, key: t.Any, value: t.Any) -> None:
-        if key in self.__slots__:
-            super(Config, self).__setattr__(key, value)
-            return
+        hidden_values = {key: "..." for key in self._schema.serialize().keys()}
+        return f"<Configuration {repr(hidden_values)}, settings_module: {self._config_module}>"
 
-        self[key] = value
+    def __str__(self) -> str:
+        return repr(self)
 
     @property
     def config_values(self) -> t.ValuesView[VT]:
         """
         Returns a copy of the dictionary of current settings.
         """
-        return super().values()
+        return self._schema.serialize().values()
+
+    def __setattr__(self, key: t.Any, value: t.Any) -> None:
+        if key in self.__slots__ + ("_initialized",):
+            # Assign to __dict__ to avoid infinite __setattr__ loops.
+            if self._initialized:
+                raise ConfigRuntimeError(
+                    f"Invalid Operation: Can not change {key} after configuration object has been created"
+                )
+
+            super().__setattr__(key, value)
+        else:
+            setattr(self._schema, key, value)
+
+    def __delattr__(self, key: t.Any) -> None:
+        if key in self.__slots__ + ("_initialized",):
+            # TODO: add test
+            raise TypeError("can't delete config attributes.")
+        delattr(self._schema, key)
+
+    def __getattr__(self, key: t.Any) -> t.Any:
+        value = getattr(self._schema, key)
+        if isinstance(value, (list, set, tuple, dict)):
+            # return immutable value
+            return type(value)(value)
+        return value
+
+    def set_defaults(self, **kwargs: t.Any) -> "Config":
+        for k, v in kwargs.items():
+            orig_value = getattr(self._schema, k, None)
+            if orig_value is None:
+                setattr(self._schema, k, v)
+        return self
+
+    def get(self, key: t.Any, _default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
+        return getattr(self, key, _default)
+
+    def __contains__(self, item: t.Any) -> bool:
+        return hasattr(self._schema, item)
+
+    def __getitem__(self, item: t.Any) -> t.Any:
+        return getattr(self, item)
+
+    def __setitem__(self, key: t.Any, value: t.Any) -> None:
+        return setattr(self, key, value)

ellar/core/conf/mixins.py

@@ -3,8 +3,10 @@
 from ellar.common.constants import LOG_LEVELS as log_levels
 from ellar.common.interfaces import IAPIVersioning, IEllarMiddleware, IExceptionHandler
 from ellar.common.templating import JinjaLoaderType
+from ellar.common.templating.validator import TemplateContextProcessorValidator
 from ellar.di import ProviderConfig
 from ellar.di.injector.tree_manager import ModuleTreeManager
+from starlette.requests import HTTPConnection, Request
 from starlette.responses import JSONResponse
 from starlette.types import ASGIApp
 
@@ -37,6 +39,13 @@ class ConfigDefaultTypesMixin:
 
     JINJA_LOADERS: t.List[t.Union[JinjaLoaderType, t.Any]]
 
+    TEMPLATES_CONTEXT_PROCESSORS: t.List[
+        t.Union[
+            TemplateContextProcessorValidator,
+            t.Callable[[t.Union[Request, HTTPConnection]], t.Dict[str, t.Any]],
+        ]
+    ]
+
     # Application route versioning scheme
     VERSIONING_SCHEME: t.Optional[IAPIVersioning]
 

ellar/core/exceptions/__init__.py

@@ -1,5 +1,21 @@
+from ellar.common import IExecutionContext
+from ellar.common.exceptions import CallableExceptionHandler
+from starlette.exceptions import HTTPException
+from starlette.responses import Response
+
 from .service import ExceptionMiddlewareService
 
 __all__ = [
     "ExceptionMiddlewareService",
+    "error_404_handler",
 ]
+
+
+async def _404_error_handler(ctx: IExecutionContext, exc: HTTPException) -> Response:
+    json_response_class = ctx.get_app().config.DEFAULT_JSON_CLASS
+    return json_response_class({"detail": exc.detail}, status_code=exc.status_code)
+
+
+error_404_handler = CallableExceptionHandler(
+    exc_or_status_code=404, handler=_404_error_handler
+)

ellar/core/exceptions/service.py

@@ -9,28 +9,30 @@
 from ellar.common.interfaces import IExceptionHandler, IExceptionMiddlewareService
 from ellar.di import injectable
 
+EXCEPTION_DEFAULT_EXCEPTION_HANDLERS: t.List[IExceptionHandler] = [
+    HTTPExceptionHandler(),
+    APIExceptionHandler(),
+    WebSocketExceptionHandler(),
+    RequestValidationErrorHandler(),
+]
+
 
 @injectable()
 class ExceptionMiddlewareService(IExceptionMiddlewareService):
-    DEFAULTS: t.List[IExceptionHandler] = [
-        HTTPExceptionHandler(),
-        APIExceptionHandler(),
-        WebSocketExceptionHandler(),
-        RequestValidationErrorHandler(),
-    ]
-
     def __init__(self) -> None:
         self._status_handlers: t.Dict[int, IExceptionHandler] = {}
         self._exception_handlers: t.Dict[t.Type[Exception], IExceptionHandler] = {}
         self._500_error_handler: t.Optional[IExceptionHandler] = None
 
-    def build_exception_handlers(self, *exception_handlers: IExceptionHandler) -> None:
-        handlers = list(self.DEFAULTS) + list(exception_handlers)
-        for key, value in handlers:
+    def build_exception_handlers(
+        self, *exception_handlers: IExceptionHandler
+    ) -> "ExceptionMiddlewareService":
+        for key, value in exception_handlers:
             if key == 500:
                 self._500_error_handler = value
                 continue
             self._add_exception_handler(key, value)
+        return self
 
     def get_500_error_handler(
         self,
@@ -58,3 +60,10 @@ def lookup_status_code_exception_handler(
         self, status_code: int
     ) -> t.Optional[IExceptionHandler]:
         return self._status_handlers.get(status_code)
+
+    def __repr__(self) -> str:  # praga: no cover
+        return (
+            f"<{self.__class__.__name__} id={id(self)}"
+            f"status-handlers={self._status_handlers} "
+            f"exc-handlers={self._exception_handlers}>"
+        )