Merge pull request #21 from eadwinCode/configuration_file_to_class

Moved Configuration From File To Class Based

Author: eadwinCode

ellar/core/conf/__init__.py

@@ -1,3 +1,4 @@
 from .config import Config
+from .mixins import ConfigDefaultTypesMixin
 
-__all__ = ["Config"]
+__all__ = ["Config", "ConfigDefaultTypesMixin"]

ellar/core/conf/app_settings_models.py

@@ -2,57 +2,15 @@
 
 from pydantic import Field, root_validator, validator
 from pydantic.json import ENCODERS_BY_TYPE
-from starlette.middleware import Middleware
 from starlette.responses import JSONResponse
 
-from ellar.core.events import EventHandler
-from ellar.core.versioning import BaseAPIVersioning, DefaultAPIVersioning
+from ellar.core.versioning import DefaultAPIVersioning
 from ellar.serializer import Serializer, SerializerFilter
 
+from .mixins import ConfigDefaultTypesMixin, TEventHandler, TMiddleware, TVersioning
 
-class TVersioning(BaseAPIVersioning):
-    @classmethod
-    def __get_validators__(
-        cls: t.Type["TVersioning"],
-    ) -> t.Iterable[t.Callable[..., t.Any]]:
-        yield cls.validate
 
-    @classmethod
-    def validate(cls: t.Type["BaseAPIVersioning"], v: t.Any) -> t.Any:
-        if not isinstance(v, BaseAPIVersioning):
-            raise ValueError(f"Expected BaseAPIVersioning, received: {type(v)}")
-        return v
-
-
-class TMiddleware(Middleware):
-    @classmethod
-    def __get_validators__(
-        cls: t.Type["TMiddleware"],
-    ) -> t.Iterable[t.Callable[..., t.Any]]:
-        yield cls.validate
-
-    @classmethod
-    def validate(cls: t.Type["Middleware"], v: t.Any) -> t.Any:
-        if not isinstance(v, Middleware):
-            raise ValueError(f"Expected Middleware, received: {type(v)}")
-        return v
-
-
-class TEventHandler(EventHandler):
-    @classmethod
-    def __get_validators__(
-        cls: t.Type["TEventHandler"],
-    ) -> t.Iterable[t.Callable[..., t.Any]]:
-        yield cls.validate
-
-    @classmethod
-    def validate(cls: t.Type["EventHandler"], v: t.Any) -> t.Any:
-        if not isinstance(v, EventHandler):
-            raise ValueError(f"Expected EventHandler, received: {type(v)}")
-        return v
-
-
-class ConfigValidationSchema(Serializer):
+class ConfigValidationSchema(Serializer, ConfigDefaultTypesMixin):
     _filter = SerializerFilter(
         exclude={
             "EXCEPTION_HANDLERS_DECORATOR",

ellar/core/conf/config.py

@@ -1,21 +1,22 @@
-import importlib
 import typing as t
 
 from starlette.config import environ
 
 from ellar.compatible.dict import AttributeDictAccessMixin, DataMutableMapper
 from ellar.constants import ELLAR_CONFIG_MODULE
+from ellar.helper.importer import import_from_string
 from ellar.types import VT
 
-from . import default_settings
 from .app_settings_models import ConfigValidationSchema
+from .default_settings import ConfigDefaults
+from .mixins import ConfigDefaultTypesMixin
 
 
 class ConfigRuntimeError(RuntimeError):
     pass
 
 
-class Config(DataMutableMapper, AttributeDictAccessMixin):
+class Config(DataMutableMapper, AttributeDictAccessMixin, ConfigDefaultTypesMixin):
     __slots__ = ("config_module", "_data")
 
     def __init__(
@@ -30,16 +31,16 @@ def __init__(
         self.config_module = config_module or environ.get(ELLAR_CONFIG_MODULE, None)
 
         self._data.clear()
-        for setting in dir(default_settings):
+        for setting in dir(ConfigDefaults):
             if setting.isupper():
-                self._data[setting] = getattr(default_settings, setting)
+                self._data[setting] = ConfigDefaults.__dict__[setting]
 
         if self.config_module:
             try:
-                mod = importlib.import_module(self.config_module)
+                mod = import_from_string(self.config_module)
                 for setting in dir(mod):
                     if setting.isupper():
-                        self._data[setting] = getattr(mod, setting)
+                        self._data[setting] = mod.__dict__[setting]
             except Exception as ex:
                 raise ConfigRuntimeError(str(ex))
 

ellar/core/conf/default_settings.py

@@ -2,7 +2,6 @@
 
 from pydantic.json import ENCODERS_BY_TYPE
 from starlette.exceptions import HTTPException as StarletteHTTPException
-from starlette.middleware import Middleware
 from starlette.responses import JSONResponse
 from starlette.types import ASGIApp
 from starlette.websockets import WebSocketClose
@@ -12,44 +11,15 @@
     request_validation_exception_handler,
 )
 from ellar.core.response import PlainTextResponse
-from ellar.core.versioning import BaseAPIVersioning, DefaultAPIVersioning
+from ellar.core.versioning import DefaultAPIVersioning
 from ellar.exceptions import APIException, RequestValidationError
 from ellar.types import TReceive, TScope, TSend
 
+from .mixins import ConfigDefaultTypesMixin, TMiddleware, TVersioning
+
 if t.TYPE_CHECKING:  # pragma: no cover
     from ellar.core.main import App
 
-DEBUG: bool = False
-
-DEFAULT_JSON_CLASS: t.Type[JSONResponse] = JSONResponse
-SECRET_KEY: str = "your-secret-key"
-
-# injector auto_bind = True allows you to resolve types that are not registered on the container
-# For more info, read: https://injector.readthedocs.io/en/latest/index.html
-INJECTOR_AUTO_BIND = False
-
-JINJA_TEMPLATES_OPTIONS: t.Dict[str, t.Any] = {}
-
-VERSIONING_SCHEME: BaseAPIVersioning = DefaultAPIVersioning()
-REDIRECT_SLASHES: bool = False
-
-STATIC_FOLDER_PACKAGES: t.Optional[t.List[t.Union[str, t.Tuple[str]]]] = []
-STATIC_DIRECTORIES: t.Optional[t.List[t.Union[str, t.Any]]] = []
-STATIC_MOUNT_PATH: str = "/static"
-
-MIDDLEWARE: t.Sequence[Middleware] = []
-
-APP_EXCEPTION_HANDLERS: t.Dict[t.Union[int, t.Type[Exception]], t.Callable] = {
-    RequestValidationError: request_validation_exception_handler,
-    APIException: api_exception_handler,
-}
-
-# 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(request, exc) -> response` and may be be either standard functions, or async functions.
-USER_CUSTOM_EXCEPTION_HANDLERS: t.Dict[t.Union[int, t.Type[Exception]], t.Callable] = {}
-
 
 async def _not_found(scope: TScope, receive: TReceive, send: TSend) -> None:
     if scope["type"] == "websocket":
@@ -67,10 +37,48 @@ async def _not_found(scope: TScope, receive: TReceive, send: TSend) -> None:
     await response(scope, receive, send)
 
 
-DEFAULT_NOT_FOUND_HANDLER: ASGIApp = _not_found
-# The lifespan context function is a newer style that replaces
-# on_startup / on_shutdown handlers. Use one or the other, not both.
-DEFAULT_LIFESPAN_HANDLER: t.Optional[t.Callable[["App"], t.AsyncContextManager]] = None
+class ConfigDefaults(ConfigDefaultTypesMixin):
+    DEBUG: bool = False
+
+    DEFAULT_JSON_CLASS: t.Type[JSONResponse] = JSONResponse
+    SECRET_KEY: str = "your-secret-key"
+
+    # injector auto_bind = True allows you to resolve types that are not registered on the container
+    # For more info, read: https://injector.readthedocs.io/en/latest/index.html
+    INJECTOR_AUTO_BIND = False
+
+    JINJA_TEMPLATES_OPTIONS: t.Dict[str, t.Any] = {}
+
+    VERSIONING_SCHEME: TVersioning = DefaultAPIVersioning()  # type: ignore
+    REDIRECT_SLASHES: bool = False
+
+    STATIC_FOLDER_PACKAGES: t.Optional[t.List[t.Union[str, t.Tuple[str, str]]]] = []
+    STATIC_DIRECTORIES: t.Optional[t.List[t.Union[str, t.Any]]] = []
+    STATIC_MOUNT_PATH: str = "/static"
+
+    MIDDLEWARE: t.List[TMiddleware] = []
+
+    APP_EXCEPTION_HANDLERS: t.Dict[t.Union[int, t.Type[Exception]], t.Callable] = {
+        RequestValidationError: request_validation_exception_handler,
+        APIException: api_exception_handler,
+    }
+
+    # 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(request, exc) -> response` and may be be either standard functions, or async functions.
+    USER_CUSTOM_EXCEPTION_HANDLERS: t.Dict[
+        t.Union[int, t.Type[Exception]], t.Callable
+    ] = {}
+
+    DEFAULT_NOT_FOUND_HANDLER: ASGIApp = _not_found
+    # The lifespan context function is a newer style that replaces
+    # on_startup / on_shutdown handlers. Use one or the other, not both.
+    DEFAULT_LIFESPAN_HANDLER: t.Optional[
+        t.Callable[["App"], t.AsyncContextManager]
+    ] = None
 
-# Object Serializer custom encoders
-SERIALIZER_CUSTOM_ENCODER: t.Dict[t.Any, t.Callable[[t.Any], t.Any]] = ENCODERS_BY_TYPE
+    # Object Serializer custom encoders
+    SERIALIZER_CUSTOM_ENCODER: t.Dict[
+        t.Any, t.Callable[[t.Any], t.Any]
+    ] = ENCODERS_BY_TYPE

ellar/core/conf/mixins.py

@@ -0,0 +1,81 @@
+import typing as t
+
+from starlette.middleware import Middleware
+from starlette.responses import JSONResponse
+
+from ellar.core.events import EventHandler
+from ellar.core.versioning import BaseAPIVersioning
+
+__all__ = ["ConfigDefaultTypesMixin", "TVersioning", "TMiddleware", "TEventHandler"]
+
+
+class TVersioning(BaseAPIVersioning):
+    @classmethod
+    def __get_validators__(
+        cls: t.Type["TVersioning"],
+    ) -> t.Iterable[t.Callable[..., t.Any]]:
+        yield cls.validate
+
+    @classmethod
+    def validate(cls: t.Type["BaseAPIVersioning"], v: t.Any) -> t.Any:
+        if not isinstance(v, BaseAPIVersioning):
+            raise ValueError(f"Expected BaseAPIVersioning, received: {type(v)}")
+        return v
+
+
+class TMiddleware(Middleware):
+    @classmethod
+    def __get_validators__(
+        cls: t.Type["TMiddleware"],
+    ) -> t.Iterable[t.Callable[..., t.Any]]:
+        yield cls.validate
+
+    @classmethod
+    def validate(cls: t.Type["Middleware"], v: t.Any) -> t.Any:
+        if not isinstance(v, Middleware):
+            raise ValueError(f"Expected Middleware, received: {type(v)}")
+        return v
+
+
+class TEventHandler(EventHandler):
+    @classmethod
+    def __get_validators__(
+        cls: t.Type["TEventHandler"],
+    ) -> t.Iterable[t.Callable[..., t.Any]]:
+        yield cls.validate
+
+    @classmethod
+    def validate(cls: t.Type["EventHandler"], v: t.Any) -> t.Any:
+        if not isinstance(v, EventHandler):
+            raise ValueError(f"Expected EventHandler, received: {type(v)}")
+        return v
+
+
+class ConfigDefaultTypesMixin:
+    DEBUG: bool
+    INJECTOR_AUTO_BIND: bool
+    DEFAULT_JSON_CLASS: t.Type[JSONResponse]
+
+    JINJA_TEMPLATES_OPTIONS: t.Dict[str, t.Any]
+    VERSIONING_SCHEME: TVersioning
+
+    REDIRECT_SLASHES: bool
+    STATIC_FOLDER_PACKAGES: t.Optional[t.List[t.Union[str, t.Tuple[str, str]]]]
+    STATIC_DIRECTORIES: t.Optional[t.List[t.Union[str, t.Any]]]
+
+    MIDDLEWARE: t.List[TMiddleware]
+    APP_EXCEPTION_HANDLERS: t.Dict[t.Union[int, t.Type[Exception]], t.Callable]
+
+    USER_CUSTOM_EXCEPTION_HANDLERS: t.Dict[t.Union[int, t.Type[Exception]], t.Callable]
+    EXCEPTION_HANDLERS: t.Dict[t.Union[int, t.Type[Exception]], t.Callable]
+    STATIC_MOUNT_PATH: str
+
+    SERIALIZER_CUSTOM_ENCODER: t.Dict[t.Any, t.Callable[[t.Any], t.Any]]
+
+    MIDDLEWARE_DECORATOR: t.List[TMiddleware]
+
+    ON_REQUEST_STARTUP: t.List[TEventHandler]
+    ON_REQUEST_SHUTDOWN: t.List[TEventHandler]
+
+    TEMPLATE_FILTERS: t.Dict[str, t.Callable[..., t.Any]]
+    TEMPLATE_GLOBAL_FILTERS: t.Dict[str, t.Callable[..., t.Any]]

ellar/core/factory.py

@@ -4,12 +4,7 @@
 
 from starlette.routing import Host, Mount
 
-from ellar.constants import (
-    MODULE_METADATA,
-    MODULE_WATERMARK,
-    ON_REQUEST_SHUTDOWN_KEY,
-    ON_REQUEST_STARTUP_KEY,
-)
+from ellar.constants import MODULE_METADATA, MODULE_WATERMARK
 from ellar.core import Config
 from ellar.core.main import App
 from ellar.core.modules import ModuleBase
@@ -71,11 +66,11 @@ def _create_app(
         assert reflect.get_metadata(MODULE_WATERMARK, module), "Only Module is allowed"
 
         config = Config(app_configured=True, config_module=config_module)
-        injector = EllarInjector(auto_bind=t.cast(bool, config.INJECTOR_AUTO_BIND))
+        injector = EllarInjector(auto_bind=config.INJECTOR_AUTO_BIND)
         cls._build_modules(app_module=module, injector=injector, config=config)
 
-        shutdown_event = config[ON_REQUEST_STARTUP_KEY]
-        startup_event = config[ON_REQUEST_SHUTDOWN_KEY]
+        shutdown_event = config.ON_REQUEST_STARTUP
+        startup_event = config.ON_REQUEST_SHUTDOWN
 
         app = App(
             config=config,

ellar/core/main.py

@@ -73,7 +73,7 @@ def __init__(
         )
         self.router = ApplicationRouter(
             routes=self._get_module_routes(),
-            redirect_slashes=t.cast(bool, self.config.REDIRECT_SLASHES),
+            redirect_slashes=self.config.REDIRECT_SLASHES,
             on_startup=[self.on_startup.async_run],
             on_shutdown=[self.on_shutdown.async_run],
             default=self.config.DEFAULT_NOT_FOUND_HANDLER,  # type: ignore

ellar/core/params/args.py

@@ -7,7 +7,7 @@
 from pydantic.error_wrappers import ErrorWrapper
 from pydantic.fields import FieldInfo, ModelField, Required
 from pydantic.schema import get_annotation_from_field_info
-from pydantic.typing import ForwardRef, evaluate_forwardref
+from pydantic.typing import ForwardRef, evaluate_forwardref  # type: ignore
 from pydantic.utils import lenient_issubclass
 from starlette.convertors import Convertor
 

ellar/core/templating/interface.py

@@ -143,7 +143,7 @@ def get_module_loaders(self) -> t.Generator[ModuleTemplating, None, None]:
 
     @property
     def debug(self) -> bool:
-        return t.cast(bool, self.config.DEBUG)
+        return self.config.DEBUG
 
     @debug.setter
     def debug(self, value: bool) -> None:
@@ -193,7 +193,7 @@ def create_global_jinja_loader(self) -> JinjaLoader:
 
     def create_static_app(self) -> ASGIApp:
         return StaticFiles(
-            directories=self.static_files, packages=self.config.STATIC_FOLDER_PACKAGES  # type: ignore
+            directories=self.static_files, packages=self.config.STATIC_FOLDER_PACKAGES
         )
 
     def reload_static_app(self) -> None: