"""
Module for loading objects defined in `settings.py` or in the Duck application configuration.
Provides functions to retrieve various components and configurations dynamically.
"""
from typing import (
Any,
List,
Dict,
Tuple,
Type,
Coroutine,
)
from duck.template.templatetags import (
TemplateTag,
TemplateFilter,
TemplateTagError,
)
from duck.exceptions.all import (
MiddlewareLoadError,
NormalizerLoadError,
SettingsError,
)
from duck.http.core.proxyhandler import HttpProxyHandler
from duck.http.request import HttpRequest
from duck.automation import Automation
from duck.automation.dispatcher import AutomationDispatcher
from duck.automation.trigger import AutomationTrigger
from duck.routes import Blueprint
from duck.html.components.core.system import LivelyComponentSystem
from duck.settings import SETTINGS
from duck.utils.importer import (import_module_once, x_import)
from duck.utils.lazy import Lazy
from duck.logging import logger
[docs]
def get_wsgi() -> Any:
"""
Returns the loaded WSGI application defined in `settings.py`.
Raises:
SettingsError: If `WSGI` is not defined or cannot be imported.
"""
wsgi_path = SETTINGS.get("WSGI")
if not wsgi_path:
raise SettingsError("Please define WSGI in `settings.py`.")
try:
return x_import(wsgi_path)(SETTINGS)
except Exception as e:
raise SettingsError(f"Failed to load WSGI: {e}") from e
[docs]
def get_asgi() -> Any:
"""
Returns the loaded ASGI application defined in `settings.py`.
Raises:
SettingsError: If `ASGI` is not defined or cannot be imported.
"""
asgi_path = SETTINGS.get("ASGI")
if not asgi_path:
raise SettingsError("Please define ASGI in `settings.py`.")
try:
return x_import(asgi_path)(SETTINGS)
except Exception as e:
raise SettingsError(f"Failed to load ASGI: {e}") from e
[docs]
def get_file_upload_handler() -> Any:
"""
Returns the file upload handler defined in `settings.py`.
Raises:
SettingsError: If `FILE_UPLOAD_HANDLER` is not defined or cannot be imported.
"""
handler_path = SETTINGS.get("FILE_UPLOAD_HANDLER")
if not handler_path:
raise SettingsError("Please define FILE_UPLOAD_HANDLER in `settings.py`.")
try:
return x_import(handler_path)
except Exception as e:
raise SettingsError(f"Failed to load file upload handler: {e}") from e
[docs]
def get_request_class() -> Type[Any]:
"""
Returns the request class defined in `settings.py`.
Raises:
SettingsError: If `REQUEST_CLASS` is not defined or cannot be imported.
"""
request_class = SETTINGS.get("REQUEST_CLASS")
if not request_class:
raise SettingsError("Please define REQUEST_CLASS in `settings.py`.")
try:
request_class = x_import(request_class)
if not issubclass(request_class, HttpRequest):
raise Exception(
"Invalid request class, should be a subclass of HttpRequest.")
return request_class
except Exception as e:
raise SettingsError(f"Failed to load request class: {e}") from e
[docs]
def get_content_compression_settings():
"""
Returns the content compression settings defined in `settings.py`.
Raises:
SettingsError: If `CONTENT_COMPRESSION` is not defined or cannot be imported.
"""
content_compression = SETTINGS.get("CONTENT_COMPRESSION")
if not content_compression:
raise SettingsError("Please define CONTENT_COMPRESSION in `settings.py`.")
try:
if not isinstance(content_compression, dict):
raise SettingsError("CONTENT_COMPRESSION should be a dictionary.")
encoding = content_compression.get("encoding")
if encoding not in ("gzip", "deflate", "br", "identity"):
raise SettingsError(
f"Invalid encoding: {encoding}. Must be one of 'gzip', 'deflate', 'identity' or 'br'."
)
if encoding == "br":
try:
pass
except ImportError as e:
raise SettingsError(
"Brotli decompression requires brotli library, please install it first using `pip install brotli`"
) from e
return content_compression
except Exception as e:
raise SettingsError(f"Failed to load content compression settings: {e}") from e
[docs]
def get_proxy_handlers() -> List[Type[Any]]:
"""
Returns loaded proxy handler classes, i.e. synchronous & asynchronous handlers..
"""
proxy_handler = SETTINGS.get("PROXY_HANDLER")
async_proxy_handler = SETTINGS.get("ASYNC_PROXY_HANDLER")
handlers = []
for proxy_handler in [proxy_handler, async_proxy_handler]:
if not proxy_handler:
if proxy_handler == async_proxy_handler:
raise SettingsError("Please define ASYNC_PROXY_HANDLER in `settings.py`.")
raise SettingsError("Please define PROXY_HANDLER in `settings.py`.")
try:
proxy_handler = x_import(proxy_handler)
if not issubclass(proxy_handler, HttpProxyHandler):
raise Exception(
"Invalid proxy handler, should be a subclass of HttpProxyHandler."
)
handlers.append(proxy_handler)
except Exception as e:
if proxy_handler == async_proxy_handler:
raise SettingsError(f"Failed to load async proxy handler: {e}") from e
raise SettingsError(f"Failed to load proxy handler: {e}") from e
return handlers
[docs]
def get_automation_dispatcher() -> AutomationDispatcher:
"""
Returns the automation dispatcher defined in `settings.py`.
Raises:
SettingsError: If `AUTOMATION_DISPATCHER` is not properly defined.
"""
dispatcher_path = SETTINGS.get("AUTOMATION_DISPATCHER")
if not dispatcher_path:
raise SettingsError(
"AUTOMATION_DISPATCHER is not set in `settings.py`. Ensure `RUN_AUTOMATIONS=True`."
)
try:
dispatcher = x_import(dispatcher_path)
if not issubclass(dispatcher, AutomationDispatcher):
raise SettingsError(
f"AUTOMATION_DISPATCHER should be a subclass of AutomationDispatcher, not {dispatcher}."
)
return dispatcher
except Exception as e:
raise SettingsError(
f"Failed to load automation dispatcher: {e}") from e
[docs]
def get_triggers_and_automations(
) -> List[Tuple[AutomationTrigger, Automation]]:
"""
Returns all triggers and their corresponding automations defined in `settings.py`.
Raises:
SettingsError: If any trigger or automation is invalid or improperly configured.
"""
automations = []
try:
for idx, (automation_path, meta) in enumerate(SETTINGS.get("AUTOMATIONS", {}).items()):
try:
automation = x_import(automation_path)
except ImportError:
raise SettingsError(
f"Cannot import automation '{automation_path}' ")
if not isinstance(automation, Automation):
raise SettingsError(
f"Automation '{automation_path}' at position {idx} must be an instance of Automation, not {type(automation).__name__}."
)
if not isinstance(meta, dict):
raise SettingsError(
f"AUTOMATIONS in settings should map Automations to dictionaries, not {type(meta).__name__}."
)
trigger_path = meta.get("trigger")
if not trigger_path:
raise SettingsError(
f"Automation '{automation_path}' at position {idx} must have a corresponding dictionary with key 'trigger' set."
)
try:
trigger = x_import(trigger_path)
except ImportError:
raise SettingsError(
f"Cannot import automation trigger '{trigger_path}' ")
if not isinstance(trigger, AutomationTrigger):
raise SettingsError(
f"Trigger for automation '{automation_path}' at position {idx} must be an instance of AutomationTrigger, not {type(trigger).__name__}."
)
try:
trigger.check_trigger() # check if this method is implemented
except NotImplementedError:
raise SettingsError("Method 'check_trigger' not implemented")
automations.append((trigger, automation))
except Exception as e:
raise SettingsError(f"Error loading automations: {e}")
return automations
[docs]
def get_blueprints() -> List[Blueprint]:
"""
Returns necessary loaded blueprints.
Notes:
- In condition that the user hasn't defined urlpatterns and blueprints,
or all of the defined blueprints are builtins,
**Duck's** default site blueprint will be added blueprints list.
"""
ducksite_blueprint = x_import("duck.etc.apps.defaultsite.blueprint.DuckSite")
blueprints = SETTINGS["BLUEPRINTS"]
final_blueprints = []
builtins_count = 0 # number of builtin blueprints
try:
for i in blueprints:
blueprint = x_import(i)
if not isinstance(blueprint, Blueprint):
raise SettingsError(
f'Blueprint "{i}" should be an instance of Blueprint not {type(blueprint)}'
)
if blueprint.is_builtin:
builtins_count += 1
final_blueprints.append(blueprint)
if SETTINGS['DEBUG'] and not SettingsLoaded.URLPATTERNS and len(blueprints) == builtins_count:
# No urlpatterns defined
# No blueprints defined or all of the defined blueprints are builtins
final_blueprints.append(ducksite_blueprint)
return final_blueprints
except Exception as e:
raise SettingsError(f"Error loading blueprints: {e}") from e
[docs]
def get_user_urlpatterns():
"""
Returns urlpatterns defined in URLPATTERNS_MODULE in settings.py
"""
try:
urlpatterns_mod = import_module_once(SETTINGS["URLPATTERNS_MODULE"])
return urlpatterns_mod.urlpatterns
except Exception as e:
raise SettingsError(f"Error loading urlpatterns: {e}") from e
[docs]
def get_user_middlewares() -> List[Type]:
"""
Loads and optionally adapts user-defined middleware classes listed in settings.
Middleware classes must define classmethods such as `process_request`,
`process_response`, and `get_error_response`.
Returns:
List[Type]: List of middleware classes, optionally patched for compatibility.
"""
try:
middlewares = [
x_import(path) for path in SETTINGS["MIDDLEWARES"]
]
return middlewares
except Exception as e:
raise MiddlewareLoadError(f"Failed to load middlewares: {e}") from e
[docs]
def get_normalizers():
"""
Returns loaded normalizers set in settings.py
"""
try:
normalizers = [
x_import(normalizer) for normalizer in SETTINGS["NORMALIZERS"]
]
return normalizers
except Exception as e:
raise NormalizerLoadError("Error loading normalizers: %s" % str(e)) from e
[docs]
def get_session_storage():
"""
Returns the session storage class defined in settings.py.
"""
return x_import(SETTINGS["SESSION_STORAGE"]) # import session storage
[docs]
def get_session_store():
"""
Returns the session store class.
"""
session_engine = SETTINGS["SESSION_ENGINE"]
return import_module_once(session_engine).SessionStore # get SessionStore from session engine.
[docs]
def get_request_handling_task_executor():
"""
Returns the request handling callable for executing
request handling threads and coroutines.
"""
try:
request_handling_task_executor = x_import(SETTINGS["REQUEST_HANDLING_TASK_EXECUTOR"])
return request_handling_task_executor()
except Exception as e:
raise SettingsError(f"Error loading request handling task executor: {e}") from e
[docs]
def get_preferred_log_style() -> str:
"""
Returns the preferred log style based on `setting.py`.
"""
style = SETTINGS["PREFERRED_LOG_STYLE"]
supported = {"duck", "django"}
if style and style not in supported:
raise SettingsError(f"PREFERRED_LOG_STYLE unsupported. Supported styles: {supported}")
[docs]
class Loaded:
"""
Loaded settings's objects.
"""
def __init__(self):
from duck.etc.templatetags import BUILTIN_TEMPLATETAGS
from duck.http.session.session_storage_connector import SessionStorageConnector
self.WSGI = Lazy(get_wsgi)
self.ASGI = Lazy(get_asgi)
self.FILE_UPLOAD_HANDLER = Lazy(get_file_upload_handler)
self.USER_TEMPLATETAGS = Lazy(get_user_templatetags)
self.BUILTIN_TEMPLATETAGS = BUILTIN_TEMPLATETAGS
self.TEMPLATE_HTML_COMPONENT_TAGS = (
LivelyComponentSystem.get_html_tags()
if SETTINGS.get("ENABLE_COMPONENT_SYSTEM")
else []
)
self.ALL_TEMPLATETAGS = (
self.USER_TEMPLATETAGS
+ self.BUILTIN_TEMPLATETAGS
+ self.TEMPLATE_HTML_COMPONENT_TAGS
)
self.REQUEST_CLASS = get_request_class()
self.CONTENT_COMPRESSION = Lazy(get_content_compression_settings)
self.PROXY_HANDLER, self.ASYNC_PROXY_HANDLER = Lazy(get_proxy_handlers) if SETTINGS["USE_DJANGO"] else (None, None)
self.AUTOMATION_DISPATCHER, self.AUTOMATIONS = (
Lazy(get_automation_dispatcher), Lazy(get_triggers_and_automations)
if SETTINGS.get("RUN_AUTOMATIONS")
else (None, []),
)
self.URLPATTERNS = Lazy(get_user_urlpatterns)
self.BLUEPRINTS = Lazy(get_blueprints)
self.MIDDLEWARES = Lazy(get_user_middlewares)
self.NORMALIZERS = Lazy(get_normalizers)
self.SESSION_STORAGE = get_session_storage()
self.SESSION_STORE = Lazy(get_session_store)
self.SESSION_STORAGE_CONNECTOR = Lazy(lambda: SessionStorageConnector(self.SESSION_STORAGE))
self.REQUEST_HANDLING_TASK_EXECUTOR = Lazy(get_request_handling_task_executor)
self.PREFERRED_LOG_STYLE = Lazy(get_preferred_log_style)
# Load middlewares instantly because it sometimes cause errors if loaded lazily in request processor
self.MIDDLEWARES() # This loads middlewares right away
self.SESSION_STORE() # Also do the same for session store.
# Initialize the loaded object.
SettingsLoaded = Lazy(Loaded) #: noqa
