WIP: dev into master #184
@@ -3,6 +3,8 @@ from typing import Callable, Self
|
||||
|
||||
from cpl.application.host import Host
|
||||
from cpl.core.console.console import Console
|
||||
from cpl.core.environment import Environment
|
||||
from cpl.core.log import LoggerABC, LogLevel
|
||||
from cpl.dependency.service_provider_abc import ServiceProviderABC
|
||||
|
||||
|
||||
@@ -38,6 +40,13 @@ class ApplicationABC(ABC):
|
||||
setattr(cls, name, func)
|
||||
return cls
|
||||
|
||||
def with_logging(self, level: LogLevel = None):
|
||||
if level is None:
|
||||
level = Environment.get("LOG_LEVEL", LogLevel, 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)
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ from socket import gethostname
|
||||
from typing import Optional, Type
|
||||
|
||||
from cpl.core.environment.environment_enum import EnvironmentEnum
|
||||
from cpl.core.typing import T
|
||||
from cpl.core.typing import T, D
|
||||
from cpl.core.utils.get_value import get_value
|
||||
|
||||
|
||||
@@ -55,14 +55,14 @@ class Environment:
|
||||
os.environ[key] = str(value)
|
||||
|
||||
@staticmethod
|
||||
def get(key: str, cast_type: Type[T], default: Optional[T] = None) -> Optional[T]:
|
||||
def get(key: str, cast_type: Type[T], default: D = None) -> T | D:
|
||||
"""
|
||||
Get an environment variable and cast it to a specified type.
|
||||
:param str key: The name of the environment variable.
|
||||
:param Type[T] cast_type: A callable to cast the variable's value.
|
||||
:param Optional[T] default: The default value to return if the variable is not found. Defaults to None.The default value to return if the variable is not found. Defaults to None.
|
||||
:param T default: The default value to return if the variable is not found. Defaults to None.The default value to return if the variable is not found. Defaults to None.
|
||||
:return: The casted value, or None if the variable is not found.
|
||||
:rtype: Optional[T]
|
||||
:rtype: T | D
|
||||
"""
|
||||
|
||||
return get_value(dict(os.environ), key, cast_type, default)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from .logger import Logger
|
||||
from .logger_abc import LoggerABC
|
||||
from .log_level_enum import LogLevelEnum
|
||||
from .log_level_enum import LogLevel
|
||||
from .logging_settings import LogSettings
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class LogLevelEnum(Enum):
|
||||
class LogLevel(Enum):
|
||||
off = "OFF" # Nothing
|
||||
trace = "TRC" # Detailed app information's
|
||||
debug = "DEB" # Detailed app state
|
||||
|
||||
@@ -3,28 +3,31 @@ import traceback
|
||||
from datetime import datetime
|
||||
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.log.log_level_enum import LogLevelEnum
|
||||
from cpl.core.log.log_level_enum import LogLevel
|
||||
from cpl.core.log.logger_abc import LoggerABC
|
||||
from cpl.core.typing import Messages, Source
|
||||
|
||||
|
||||
class Logger(LoggerABC):
|
||||
_level = LogLevelEnum.info
|
||||
_levels = [x for x in LogLevelEnum]
|
||||
_level = LogLevel.info
|
||||
_levels = [x for x in LogLevel]
|
||||
|
||||
# ANSI color codes for different log levels
|
||||
_COLORS = {
|
||||
LogLevelEnum.trace: "\033[37m", # Light Gray
|
||||
LogLevelEnum.debug: "\033[94m", # Blue
|
||||
LogLevelEnum.info: "\033[92m", # Green
|
||||
LogLevelEnum.warning: "\033[93m", # Yellow
|
||||
LogLevelEnum.error: "\033[91m", # Red
|
||||
LogLevelEnum.fatal: "\033[95m", # Magenta
|
||||
LogLevel.trace: "\033[37m", # Light Gray
|
||||
LogLevel.debug: "\033[94m", # Blue
|
||||
LogLevel.info: "\033[92m", # Green
|
||||
LogLevel.warning: "\033[93m", # Yellow
|
||||
LogLevel.error: "\033[91m", # Red
|
||||
LogLevel.fatal: "\033[95m", # Magenta
|
||||
}
|
||||
|
||||
def __init__(self, source: Source, file_prefix: str = None):
|
||||
LoggerABC.__init__(self)
|
||||
assert source is not None and source != "", "Source cannot be None or empty"
|
||||
|
||||
if source == LoggerABC.__name__:
|
||||
source = None
|
||||
|
||||
self._source = source
|
||||
|
||||
if file_prefix is None:
|
||||
@@ -45,7 +48,7 @@ class Logger(LoggerABC):
|
||||
os.makedirs("logs")
|
||||
|
||||
@classmethod
|
||||
def set_level(cls, level: LogLevelEnum):
|
||||
def set_level(cls, level: LogLevel):
|
||||
if level in cls._levels:
|
||||
cls._level = level
|
||||
else:
|
||||
@@ -69,7 +72,7 @@ class Logger(LoggerABC):
|
||||
log_file.write(content + "\n")
|
||||
log_file.close()
|
||||
|
||||
def _log(self, level: LogLevelEnum, *messages: Messages):
|
||||
def _log(self, level: LogLevel, *messages: Messages):
|
||||
try:
|
||||
if self._levels.index(level) < self._levels.index(self._level):
|
||||
return
|
||||
@@ -78,7 +81,7 @@ class Logger(LoggerABC):
|
||||
formatted_message = self._format_message(level.value, timestamp, *messages)
|
||||
|
||||
self._write_log_to_file(formatted_message)
|
||||
Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{formatted_message}\033[0m")
|
||||
Console.write_line(f"{self._COLORS.get(level, '\033[0m')}{formatted_message}\033[0m")
|
||||
except Exception as e:
|
||||
print(f"Error while logging: {e} -> {traceback.format_exc()}")
|
||||
|
||||
@@ -91,27 +94,35 @@ class Logger(LoggerABC):
|
||||
|
||||
messages = [str(message) for message in messages if message is not None]
|
||||
|
||||
return f"<{timestamp}> [{level.upper():^3}] [{self._file_prefix}] - [{self._source}]: {' '.join(messages)}"
|
||||
message = f"<{timestamp}>"
|
||||
message += f" [{level.upper():^3}]"
|
||||
message += f" [{self._file_prefix}]"
|
||||
if self._source is not None:
|
||||
message += f" - [{self._source}]"
|
||||
|
||||
message += f": {' '.join(messages)}"
|
||||
|
||||
return message
|
||||
|
||||
def header(self, string: str):
|
||||
self._log(LogLevelEnum.info, string)
|
||||
self._log(LogLevel.info, string)
|
||||
|
||||
def trace(self, *messages: Messages):
|
||||
self._log(LogLevelEnum.trace, *messages)
|
||||
self._log(LogLevel.trace, *messages)
|
||||
|
||||
def debug(self, *messages: Messages):
|
||||
self._log(LogLevelEnum.debug, *messages)
|
||||
self._log(LogLevel.debug, *messages)
|
||||
|
||||
def info(self, *messages: Messages):
|
||||
self._log(LogLevelEnum.info, *messages)
|
||||
self._log(LogLevel.info, *messages)
|
||||
|
||||
def warning(self, *messages: Messages):
|
||||
self._log(LogLevelEnum.warning, *messages)
|
||||
self._log(LogLevel.warning, *messages)
|
||||
|
||||
def error(self, message, e: Exception = None):
|
||||
self._log(LogLevelEnum.error, message, f"{e} -> {traceback.format_exc()}" if e else None)
|
||||
self._log(LogLevel.error, message, f"{e} -> {traceback.format_exc()}" if e else None)
|
||||
|
||||
def fatal(self, message, e: Exception = None, prevent_quit: bool = False):
|
||||
self._log(LogLevelEnum.fatal, message, f"{e} -> {traceback.format_exc()}" if e else None)
|
||||
self._log(LogLevel.fatal, message, f"{e} -> {traceback.format_exc()}" if e else None)
|
||||
if not prevent_quit:
|
||||
exit(-1)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
from abc import abstractmethod, ABC
|
||||
|
||||
from cpl.core.log.log_level_enum import LogLevel
|
||||
from cpl.core.typing import Messages
|
||||
|
||||
|
||||
@@ -7,7 +8,7 @@ class LoggerABC(ABC):
|
||||
r"""ABC for :class:`cpl.core.log.logger_service.Logger`"""
|
||||
|
||||
@abstractmethod
|
||||
def set_level(self, level: str): ...
|
||||
def set_level(self, level: LogLevel): ...
|
||||
|
||||
@abstractmethod
|
||||
def _format_message(self, level: str, timestamp, *messages: Messages) -> str: ...
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
from typing import Optional
|
||||
|
||||
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
|
||||
from cpl.core.log.log_level_enum import LogLevelEnum
|
||||
from cpl.core.log.log_level_enum import LogLevel
|
||||
|
||||
|
||||
class LogSettings(ConfigurationModelABC):
|
||||
@@ -11,14 +11,14 @@ class LogSettings(ConfigurationModelABC):
|
||||
self,
|
||||
path: str = None,
|
||||
filename: str = None,
|
||||
console_log_level: LogLevelEnum = None,
|
||||
file_log_level: LogLevelEnum = None,
|
||||
console_log_level: LogLevel = None,
|
||||
file_log_level: LogLevel = None,
|
||||
):
|
||||
ConfigurationModelABC.__init__(self)
|
||||
self._path: Optional[str] = path
|
||||
self._filename: Optional[str] = filename
|
||||
self._console: Optional[LogLevelEnum] = console_log_level
|
||||
self._level: Optional[LogLevelEnum] = file_log_level
|
||||
self._console: Optional[LogLevel] = console_log_level
|
||||
self._level: Optional[LogLevel] = file_log_level
|
||||
|
||||
@property
|
||||
def path(self) -> str:
|
||||
@@ -37,17 +37,17 @@ class LogSettings(ConfigurationModelABC):
|
||||
self._filename = filename
|
||||
|
||||
@property
|
||||
def console(self) -> LogLevelEnum:
|
||||
def console(self) -> LogLevel:
|
||||
return self._console
|
||||
|
||||
@console.setter
|
||||
def console(self, console: LogLevelEnum) -> None:
|
||||
def console(self, console: LogLevel) -> None:
|
||||
self._console = console
|
||||
|
||||
@property
|
||||
def level(self) -> LogLevelEnum:
|
||||
def level(self) -> LogLevel:
|
||||
return self._level
|
||||
|
||||
@level.setter
|
||||
def level(self, level: LogLevelEnum) -> None:
|
||||
def level(self, level: LogLevel) -> None:
|
||||
self._level = level
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
from enum import Enum
|
||||
from typing import Type, Optional
|
||||
|
||||
from cpl.core.typing import T
|
||||
@@ -40,6 +41,19 @@ def get_value(
|
||||
if cast_type == bool:
|
||||
return value.lower() in ["true", "1"]
|
||||
|
||||
if issubclass(cast_type, Enum):
|
||||
try:
|
||||
return cast_type(value)
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
try:
|
||||
return cast_type[value]
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
return default
|
||||
|
||||
if (cast_type if not hasattr(cast_type, "__origin__") else cast_type.__origin__) == list:
|
||||
if not (value.startswith("[") and value.endswith("]")) and list_delimiter not in value:
|
||||
raise ValueError("List values must be enclosed in square brackets or use a delimiter.")
|
||||
|
||||
@@ -14,7 +14,7 @@ class Application(ApplicationABC):
|
||||
def __init__(self, services: ServiceProviderABC):
|
||||
ApplicationABC.__init__(self, services)
|
||||
|
||||
self._logger: LoggerABC = services.get_service(LoggerABC)
|
||||
self._logger = services.get_service(LoggerABC)
|
||||
|
||||
async def test_daos(self):
|
||||
userDao: UserDao = self._services.get_service(UserDao)
|
||||
|
||||
@@ -8,6 +8,7 @@ def main():
|
||||
builder = ApplicationBuilder(Application).with_startup(Startup)
|
||||
app = builder.build()
|
||||
|
||||
app.with_logging()
|
||||
app.with_permissions(CustomPermissions)
|
||||
app.with_migrations("./scripts")
|
||||
app.with_seeders()
|
||||
|
||||
@@ -2,14 +2,18 @@ from application import Application
|
||||
from cpl.application import ApplicationBuilder
|
||||
from cpl.auth.permission.permissions_registry import PermissionsRegistry
|
||||
from cpl.core.console import Console
|
||||
from cpl.core.log import LogLevel
|
||||
from custom_permissions import CustomPermissions
|
||||
from startup import Startup
|
||||
|
||||
|
||||
def main():
|
||||
builder = ApplicationBuilder(Application).with_startup(Startup)
|
||||
builder.services.add_logging()
|
||||
|
||||
app = builder.build()
|
||||
|
||||
app.with_logging(LogLevel.trace)
|
||||
app.with_permissions(CustomPermissions)
|
||||
app.with_migrations("./scripts")
|
||||
app.with_seeders()
|
||||
|
||||
Reference in New Issue
Block a user