Better modules
All checks were successful
Test before pr merge / test-lint (pull_request) Successful in 6s
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 18s
Build on push / query (push) Successful in 19s
Build on push / dependency (push) Successful in 17s
Build on push / mail (push) Successful in 16s
Build on push / translation (push) Successful in 16s
Build on push / application (push) Successful in 18s
Build on push / database (push) Successful in 19s
Build on push / auth (push) Successful in 17s
Build on push / api (push) Successful in 14s

This commit is contained in:
2025-09-26 12:55:00 +02:00
parent 56a16cbeba
commit c410a692be
29 changed files with 341 additions and 223 deletions

View File

@@ -2,10 +2,12 @@ from abc import ABC, abstractmethod
from typing import Callable, Self
from cpl.application.host import Host
from cpl.core.errors import module_dependency_error
from cpl.core.log.log_level import LogLevel
from cpl.core.log.log_settings import LogSettings
from cpl.core.log.logger_abc import LoggerABC
from cpl.dependency.service_provider import ServiceProvider
from cpl.dependency.typing import TModule
def __not_implemented__(package: str, func: Callable):
@@ -20,17 +22,6 @@ class ApplicationABC(ABC):
Contains instances of prepared objects
"""
@abstractmethod
def __init__(self, services: ServiceProvider, required_modules: list[str | object] = None):
self._services = services
self._required_modules = (
[x.__name__ if not isinstance(x, str) else x for x in required_modules] if required_modules else []
)
@property
def required_modules(self) -> list[str]:
return self._required_modules
@classmethod
def extend(cls, name: str | Callable, func: Callable[[Self], Self]):
r"""Extend the Application with a custom method
@@ -47,6 +38,30 @@ class ApplicationABC(ABC):
setattr(cls, name, func)
return cls
@abstractmethod
def __init__(
self, services: ServiceProvider, loaded_modules: set[TModule], required_modules: list[str | object] = None
):
self._services = services
self._modules = loaded_modules
self._required_modules = (
[x.__name__ if not isinstance(x, str) else x for x in required_modules] if required_modules else []
)
def validate_app_required_modules(self):
modules_names = {x.__name__ for x in self._modules}
for module in self._required_modules:
if module in modules_names:
continue
module_dependency_error(
type(self).__name__,
module.__name__,
ImportError(
f"Required module '{module}' for application '{self.__class__.__name__}' is not loaded. Load using 'add_module({module})' method."
),
)
def with_logging(self, level: LogLevel = None):
if level is None:
from cpl.core.configuration.configuration import Configuration
@@ -57,14 +72,21 @@ class ApplicationABC(ABC):
logger = self._services.get_service(LoggerABC)
logger.set_level(level)
def with_permissions(self, *args, **kwargs):
__not_implemented__("cpl-auth", self.with_permissions)
def with_permissions(self, *args):
try:
from cpl.auth import AuthModule
def with_migrations(self, *args, **kwargs):
__not_implemented__("cpl-database", self.with_migrations)
AuthModule.with_permissions(*args)
except ImportError:
__not_implemented__("cpl-auth", self.with_permissions)
def with_seeders(self, *args, **kwargs):
__not_implemented__("cpl-database", self.with_seeders)
def with_migrations(self, *args):
try:
from cpl.database.database_module import DatabaseModule
DatabaseModule.with_migrations(self._services, *args)
except ImportError:
__not_implemented__("cpl-database", self.with_migrations)
def with_extension(self, func: Callable[[Self, ...], None], *args, **kwargs):
r"""Extend the Application with a custom method
@@ -84,6 +106,11 @@ class ApplicationABC(ABC):
Called by custom Application.main
"""
try:
for module in self._modules:
if not hasattr(module, "configure") and not callable(getattr(module, "configure")):
continue
module.configure(self._services)
Host.run_app(self.main)
except KeyboardInterrupt:
pass

View File

@@ -43,19 +43,6 @@ class ApplicationBuilder(Generic[TApp]):
return provider
def validate_app_required_modules(self, app: ApplicationABC):
for module in app.required_modules:
if module in self._services.loaded_modules:
continue
dependency_error(
type(app).__name__,
module,
ImportError(
f"Required module '{module}' for application '{app.__class__.__name__}' is not loaded. Load using 'add_module({module})' method."
),
)
def with_startup(self, startup: Type[StartupABC]) -> "ApplicationBuilder":
self._startup = startup
return self
@@ -84,6 +71,6 @@ class ApplicationBuilder(Generic[TApp]):
Host.run(extension.run, self.service_provider)
use_root_provider(self._services.build())
app = self._app(self.service_provider)
self.validate_app_required_modules(app)
app = self._app(self.service_provider, self._services.loaded_modules)
app.validate_app_required_modules()
return app