Coverage for lynceus/core/lynceus.py: 94%
88 statements
« prev ^ index » next coverage.py v7.10.0, created at 2025-07-29 08:46 +0000
1import configparser
2import logging.config
3import logging.handlers
4from logging import Logger
5from pathlib import Path
7from lynceus.core.config import CONFIG_GENERAL_KEY
8from lynceus.core.config.lynceus_config import LynceusConfig
9from lynceus.lynceus_exceptions import LynceusConfigError
10from lynceus.utils import format_exception_human_readable, lookup_root_path
13class LynceusSession:
14 """
15 Central session management class for the Lynceus framework.
17 This class manages configuration loading, logging setup, and session lifecycle
18 for Lynceus applications. It implements a singleton-like pattern using registration
19 keys to ensure proper session management and prevent memory leaks.
21 The class handles:
22 - Configuration file loading (default and user-specific)
23 - Logger initialization and management
24 - Session registration and retrieval
25 - Fine-tuned logging configuration
27 Attributes:
28 TRACE (int): Additional log level more verbose than DEBUG
29 """
30 DEFAULT_SALT: str = 'lynceus'
32 # Additional even more verbose than DEBUG log level.
33 TRACE = 5
35 __internal_key_preventing_external_usage_of_init = object()
37 __REGISTERED_SESSIONS: dict = {}
39 def __init__(self, *,
40 salt: str = DEFAULT_SALT,
41 root_logger_name: str = 'Lynceus',
42 load_core_default_config: bool = True,
43 overridden_root_logger_level: int | str | None = None,
44 _creation_key=None):
45 """
46 Initialize a new Lynceus session with configuration and logging setup.
48 **WARNING**: This constructor should NEVER be called directly by clients.
49 Always use :py:meth:`~lynceus.core.lynceus.LynceusSession.get_session`
50 to obtain session instances.
52 This method sets up the complete session infrastructure including:
53 - Configuration file loading (default and user-specific)
54 - Logger initialization with custom TRACE level
55 - Root logger configuration with specified level
57 Parameters
58 ----------
59 salt : str, optional
60 Unique identifier for this session, used for configuration file lookup
61 root_logger_name : str, optional
62 Name prefix for all loggers created by this session.
63 Used as namespace for :py:meth:`~lynceus.core.lynceus.LynceusSession.get_logger`
64 load_core_default_config : bool, optional
65 Whether to load Lynceus framework default configuration.
66 Strongly recommended for proper framework operation
67 overridden_root_logger_level : int, str, or None, optional
68 Override for root logger level from configuration.
69 Useful for CLI tools with --quiet, --debug options
70 _creation_key : optional
71 Internal safeguard key to prevent direct constructor usage.
72 Should NEVER be specified by external callers
74 Raises
75 ------
76 RuntimeError
77 If called directly without proper internal creation key
79 Notes
80 -----
81 - Automatically registers custom TRACE logging level (more verbose than DEBUG)
82 - Loads configuration files in order: default → salt-specific (default and user) → salt-flavor-specific (according to merged configuration) → user files
83 - Sets up root logger with level from config or override parameter
84 """
85 # Safe-guard: prevents direct call to this method.
86 if _creation_key != LynceusSession.__internal_key_preventing_external_usage_of_init:
87 raise RuntimeError(f'You should always use LynceusSession.get_session() method when requesting a LynceusSession ({salt=}).')
89 # Initializes internal Lynceus config.
90 self.__config = LynceusConfig()
91 self.__root_logger_name = root_logger_name
93 # Loads configuration file(s):
94 # - first the default config according to load_core_default_config toggle
95 # - then the default and user config corresponding to specify salt, if not the default one
96 loaded_config_files = self.__load_configuration(salt, load_core_default_config=load_core_default_config)
98 # Additional even more verbose than DEBUG log level.
99 logging.TRACE = LynceusSession.TRACE
100 logging.addLevelName(LynceusSession.TRACE, 'TRACE')
102 # Setups default/root Lynceus logger.
103 self.__logger = self.get_logger()
104 root_level: int | str = logging.getLevelName(overridden_root_logger_level or self.get_config(CONFIG_GENERAL_KEY, 'logger.root.level', default='INFO'))
105 self.__logger.setLevel(root_level)
107 # Informs now that Logger is initialized.
108 self.__logger.info(f'[{salt=}]This is the loading information of looked up configuration files (Format=<Path: isLoaded?>) "{loaded_config_files}".')
110 @staticmethod
111 # pylint: disable=protected-access
112 def get_session(*,
113 salt: str = DEFAULT_SALT,
114 registration_key: dict[str, str] | None = None,
115 root_logger_name: str = 'Lynceus',
116 load_core_default_config: bool = True,
117 overridden_root_logger_level: str | None = None) -> 'LynceusSession':
118 """
119 Get or create a Lynceus session with specified configuration.
121 This is the main entry point for obtaining a LynceusSession instance.
122 It implements session caching based on registration keys to prevent
123 duplicate sessions and memory leaks.
125 Parameters
126 ----------
127 salt : str, optional
128 Unique identifier for the session (default: 'lynceus')
129 registration_key : dict or None, optional
130 Dictionary used to identify and cache sessions.
131 Required for non-default salt values to prevent memory leaks.
132 root_logger_name : str, optional
133 Name prefix for all loggers created by this session
134 load_core_default_config : bool, optional
135 Whether to load Lynceus default configuration
136 overridden_root_logger_level : str or None, optional
137 Override the root logger level from config
139 Returns
140 -------
141 LynceusSession
142 A configured Lynceus session instance
144 Raises
145 ------
146 LynceusConfigError
147 If registration_key is missing for non-default salt
148 """
149 # Safe-guard: a registration key is highly recommended if not an internal session (which would be the case with salt == DEFAULT_SALT).
150 # Important: sometimes caller uses an optional **kwargs which then leads to empty dict (which is not None), and thus must be taken care too.
151 if not registration_key:
152 if salt != LynceusSession.DEFAULT_SALT:
153 raise LynceusConfigError('It is mandatory to provide your own registration_key when getting a session.' +
154 f' System will automatically consider it as {registration_key}, but you may not be able to retrieve your session.' +
155 ' And it may lead to memory leak.')
156 registration_key = {'default': salt}
158 # Turns the registration_key to an hashable version.
159 if registration_key is not None:
160 registration_key = frozenset(registration_key.items())
162 # Creates and registers the session if needed.
163 if registration_key in LynceusSession.__REGISTERED_SESSIONS:
164 session: LynceusSession = LynceusSession.__REGISTERED_SESSIONS[registration_key]
165 session.__logger.debug(f'Successfully retrieved cached Session for registration key "{registration_key}": {session}.')
166 else:
167 session: LynceusSession = LynceusSession(salt=salt, root_logger_name=root_logger_name,
168 load_core_default_config=load_core_default_config,
169 overridden_root_logger_level=overridden_root_logger_level,
170 _creation_key=LynceusSession.__internal_key_preventing_external_usage_of_init)
172 LynceusSession.__REGISTERED_SESSIONS[registration_key] = session
173 session.__logger.debug(f'Successfully registered new Session (new total count={len(LynceusSession.__REGISTERED_SESSIONS)})'
174 f' for registration key "{registration_key}": {session}.'
175 f' This is its loaded configuration (salt={salt}):\n' +
176 f'{LynceusConfig.format_dict_to_string(LynceusConfig.format_config(session.__config), indentation_level=1)}')
178 # TODO: implement another method allowing to free a specific registered session
179 # TODO: turn LynceusSession as a Context, allowing usage with the 'with' keyword to automatic free the session once used
181 # Returns the Lynceus session.
182 return session
184 def __do_load_configuration_files(self, config_file_meta_list) -> dict[Path, bool]:
185 """
186 Load configuration files from a list of metadata specifications.
188 This method processes a list of configuration file metadata and attempts
189 to load each file into the internal configuration. It handles missing files
190 gracefully and sets up fine-tuned logging configurations when present.
192 Parameters
193 ----------
194 config_file_meta_list : list
195 List of dictionaries containing configuration file metadata with keys:
196 - 'config_path': Path to the configuration file
197 - 'root_path': Optional root path for file lookup
199 Returns
200 -------
201 dict[Path, bool]
202 Mapping of configuration file paths to loading status
204 Notes
205 -----
206 Missing configuration files are handled gracefully and logged as debug messages.
207 Files with logger sections trigger fine-tuned logging configuration.
208 """
209 loaded_config_files: dict[Path, bool] = {}
210 for config_file_meta in config_file_meta_list:
211 conf_file_name = config_file_meta['config_path']
212 conf_file_root_path = config_file_meta.get('root_path', Path().resolve())
214 loaded: bool = False
215 relative_file: Path = Path(conf_file_name)
216 try:
217 full_path: Path = lookup_root_path(relative_file, root_path=conf_file_root_path) / relative_file
219 # Merges it in internal Lynceus config.
220 additional_config: configparser.ConfigParser = self.__config.update_from_configuration_file(full_path)
222 # Updates/Configures fine-tuned logger, if any in this (first or) additional configuration file.
223 # N.B.: in ideal world, could be interesting to extract logger config from all configuration file(s) merged, and
224 # create a fake fileConfig in memory, to logging.config.fileConfig once for all, after all loading.
225 # Without that, handlers and formatters must be redefined each time, in each configuration file definined fine-tuned logger configuration.
226 if 'loggers' in additional_config.sections():
227 self.get_logger('internal').debug(f'Requesting fine-tuned logger configuration with configuration coming from {conf_file_name}.')
228 # Cf. https://docs.python.org/3.10/library/logging.config.html#logging.config.fileConfig
229 logging.config.fileConfig(full_path, disable_existing_loggers=False, encoding='utf-8')
231 loaded = True
232 except FileNotFoundError as exc:
233 # Since recent version, there is no more mandatory configuration file to have, and several files can be loaded.
234 # N.B.: it is valid, there is no logger yet here, so using print ...
235 self.get_logger('internal').debug(f'WARNING: Unable to load "{conf_file_name}" configuration file,'
236 f' from "{conf_file_root_path}"'
237 f' (it is OK, because this configuration file is NOT mandatory) =>'
238 f' {format_exception_human_readable(exc, quote_message=True)}')
240 # Registers configuration file loading information.
241 loaded_config_files[relative_file] = loaded
242 return loaded_config_files
244 def __load_configuration(self, salt: str, load_core_default_config: bool) -> dict[Path, bool]:
245 """
246 Load all relevant configuration files for the session.
248 This method orchestrates the loading of multiple configuration files
249 in a specific order:
250 1. Lynceus default configuration (if requested)
251 2. Salt-specific default configuration
252 3. Salt-specific configuration file
253 4. Additional Salt-specific and flavor-specific configuration files
255 Parameters
256 ----------
257 salt : str
258 Session identifier used to determine configuration file names
259 load_core_default_config : bool
260 Whether to load the Lynceus default config
262 Returns
263 -------
264 dict[Path, bool]
265 Mapping of all configuration file paths to their loading status
267 Notes
268 -----
269 Configuration files are loaded in order of precedence, with later files
270 potentially overriding settings from earlier ones.
271 """
272 # Defines the potential configuration file to load.
273 config_file_meta_list = []
274 if load_core_default_config:
275 # Adds the Lynceus default configuration file as requested.
276 config_file_meta_list.append({'config_path': f'misc/{self.DEFAULT_SALT}.default.conf', 'root_path': Path(__file__).parent})
278 # Adds default configuration file corresponding to specified salt, if not the default one.
279 if salt != self.DEFAULT_SALT:
280 config_file_meta_list.append({'config_path': f'misc/{salt}.default.conf'})
282 # Adds user configuration file corresponding to specified salt.
283 config_file_meta_list.append({'config_path': f'{salt}.conf'})
285 # Looks up for configuration file(s), starting with a "default" one, and then optional user one.
286 loaded_config_files: dict[Path, bool] = self.__do_load_configuration_files(config_file_meta_list)
288 # Loads additional configuration files based on the optional conf_flavor_list setting, which can be specified in any of the configuration files that have just been loaded.
289 conf_flavor_list: list[str] = self.get_config(CONFIG_GENERAL_KEY, 'conf_flavor_list', default=[])
290 if conf_flavor_list:
291 extra_config_file_meta_list = [{'config_path': f'{salt}_{flavor}.conf'} for flavor in conf_flavor_list]
292 loaded_config_files |= self.__do_load_configuration_files(extra_config_file_meta_list)
294 # Returns configuration file loading mapping, for information.
295 return loaded_config_files
297 def has_config_section(self, section: str) -> bool:
298 """
299 Check if a configuration section exists.
301 Parameters
302 ----------
303 section : str
304 Name of the configuration section to check
306 Returns
307 -------
308 bool
309 True if the section exists, False otherwise
310 """
311 return self.__config.has_section(section)
313 def get_config(self, section: str, key: str, *, default: str | int | float | object | list = LynceusConfig.UNDEFINED_VALUE) -> str | int | float | object | list:
314 """
315 Retrieve a configuration value from the specified section and key.
317 Parameters
318 ----------
319 section : str
320 Configuration section name
321 key : str
322 Configuration key within the section
323 default : str or int or float or object or list, optional
324 Default value to return if the config doesn't exist.
326 Returns
327 -------
328 str or int or float or object or list
329 The configuration value or default.
331 Raises
332 ------
333 LynceusConfigError
334 If the key is not found and no default is provided
335 """
336 return self.__config.get_config(section, key, default=default)
338 def is_bool_config_enabled(self, section: str, key: str) -> bool:
339 """
340 Check if a boolean configuration option is enabled.
342 Parameters
343 ----------
344 section : str
345 Configuration section name
346 key : str
347 Configuration key within the section
349 Returns
350 -------
351 bool
352 True if the configuration value evaluates to True, False otherwise
353 """
354 return self.__config.is_bool_config_enabled(section=section, key=key)
356 def get_config_section(self, section: str):
357 """
358 Retrieve an entire configuration section.
360 Parameters
361 ----------
362 section : str
363 Name of the configuration section to retrieve
365 Returns
366 -------
367 dict
368 Configuration section object containing all key-value pairs
370 Raises
371 ------
372 KeyError
373 If the section does not exist
374 """
375 return self.__config[section]
377 def get_lynceus_config_copy(self) -> LynceusConfig:
378 """
379 Create a deep copy of the session's internal configuration.
381 This method returns a complete copy of the current configuration state,
382 including all loaded configuration files and merged settings. The copy
383 is independent of the original and can be safely modified without
384 affecting the session's configuration.
386 Returns
387 -------
388 LynceusConfig
389 Deep copy of the internal configuration with all
390 loaded settings from default, user, and flavor-specific
391 configuration files
393 Notes
394 -----
395 - Useful for creating configuration templates or snapshots
396 - The returned copy includes all merged configuration from multiple sources
397 - Changes to the returned copy do not affect the session's configuration
398 - Can be used to extract default configuration values for documentation
399 """
400 return self.__config.copy()
402 def get_logger(self, name: str = None, *, parent_propagate: bool = True) -> Logger:
403 """
404 Get a logger instance with the session's root logger as prefix.
406 Create or retrieve a logger with the full name constructed from
407 the session's root logger name and the provided name.
409 Parameters
410 ----------
411 name : str, optional
412 Specific logger name (optional). If None, returns the root logger
413 parent_propagate : bool, optional
414 Whether the logger should propagate to its parent logger
416 Returns
417 -------
418 Logger
419 Configured logger instance
420 """
421 complete_name: str = f'{self.__root_logger_name}.{name}' if name else self.__root_logger_name
422 logger: Logger = logging.getLogger(complete_name)
423 if logger.parent:
424 logger.parent.propagate = parent_propagate
426 return logger