edraft
12b7c62b69
All checks were successful
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 17s
Build on push / query (push) Successful in 17s
Build on push / dependency (push) Successful in 17s
Build on push / translation (push) Successful in 14s
Build on push / application (push) Successful in 18s
Build on push / database (push) Successful in 17s
Build on push / mail (push) Successful in 18s
Build on push / auth (push) Successful in 13s
Build on push / api (push) Successful in 17s
Test before pr merge / test-lint (pull_request) Successful in 5s
94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
from abc import ABC, abstractmethod
|
|
from typing import Callable, Self
|
|
|
|
from cpl.application.host import Host
|
|
from cpl.core.console.console import Console
|
|
from cpl.core.log import LogSettings
|
|
from cpl.core.log.log_level import LogLevel
|
|
from cpl.core.log.logger_abc import LoggerABC
|
|
from cpl.dependency.service_provider_abc import ServiceProviderABC
|
|
|
|
|
|
def __not_implemented__(package: str, func: Callable):
|
|
raise NotImplementedError(f"Package {package} is required to use {func.__name__} method")
|
|
|
|
|
|
class ApplicationABC(ABC):
|
|
r"""ABC for the Application class
|
|
|
|
Parameters:
|
|
services: :class:`cpl.dependency.service_provider_abc.ServiceProviderABC`
|
|
Contains instances of prepared objects
|
|
"""
|
|
|
|
@abstractmethod
|
|
def __init__(self, services: ServiceProviderABC, 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
|
|
|
|
Parameters:
|
|
name: :class:`str`
|
|
Name of the method
|
|
func: :class:`Callable[[Self], Self]`
|
|
Function that takes the Application as a parameter and returns it
|
|
"""
|
|
if callable(name):
|
|
name = name.__name__
|
|
|
|
setattr(cls, name, func)
|
|
return cls
|
|
|
|
def with_logging(self, level: LogLevel = None):
|
|
if level is None:
|
|
from cpl.core.configuration.configuration import Configuration
|
|
|
|
settings = Configuration.get(LogSettings)
|
|
level = settings.level if settings else LogLevel.info
|
|
|
|
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_migrations(self, *args, **kwargs):
|
|
__not_implemented__("cpl-database", self.with_migrations)
|
|
|
|
def with_seeders(self, *args, **kwargs):
|
|
__not_implemented__("cpl-database", self.with_seeders)
|
|
|
|
def with_extension(self, func: Callable[[Self, ...], None], *args, **kwargs):
|
|
r"""Extend the Application with a custom method
|
|
|
|
Parameters:
|
|
func: :class:`Callable[[Self], Self]`
|
|
Function that takes the Application as a parameter and returns it
|
|
"""
|
|
assert func is not None, "func must not be None"
|
|
assert callable(func), "func must be callable"
|
|
|
|
func(self, *args, **kwargs)
|
|
|
|
def run(self):
|
|
r"""Entry point
|
|
|
|
Called by custom Application.main
|
|
"""
|
|
try:
|
|
Host.run(self.main)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def main(self): ...
|