Coverage for lynceus/core/lynceus_client.py: 89%
72 statements
« prev ^ index » next coverage.py v7.10.0, created at 2025-07-29 08:46 +0000
1import logging
2import time
3from abc import ABCMeta
4from datetime import datetime, timezone
5from logging import CRITICAL, DEBUG, ERROR, INFO, Logger, WARNING
7from .config import CONFIG_LOG_MESSAGE_STATUS_KEY
8from .exchange.lynceus_exchange import LynceusExchange
9from .lynceus import LynceusSession
10from .lynceus_message_status import LynceusMessageStatus
13class LynceusClientLogExchangeWrapper(Logger):
14 """
15 Logger wrapper that integrates with LynceusExchange for message queuing.
16 """
17 # Defines a threshold for automatic message status definition (can be: 0, WARNING, or ERROR).
18 AUTOMATIC_MESSAGE_STATUS_LOG_LEVEL_THRESHOLD: int = ERROR
20 # N.B.: Logger has already been initialized, and specified as argument, so we do NOT want to call super __init__ method.
21 # pylint: disable=super-init-not-called
22 def __init__(self, logger: Logger, lynceus_exchange: LynceusExchange | None):
23 """
24 Initialize the logger wrapper.
26 Parameters
27 ----------
28 logger : Logger
29 The underlying Logger instance to wrap.
30 lynceus_exchange : LynceusExchange or None
31 Optional exchange for message queuing.
32 """
33 self.__logger: Logger = logger
34 self._lynceus_exchange: LynceusExchange | None = lynceus_exchange
36 # Adds various properties wrapping around internal logger property allowing to use custom Level like TRACE.
37 @property
38 def name(self) -> str:
39 """
40 Get the logger name.
42 Returns
43 -------
44 str
45 The logger name.
46 """
47 return self.__logger.name
49 @property
50 def level(self) -> int:
51 """
52 Get the logger level.
54 Returns
55 -------
56 int
57 The logger level.
58 """
59 return self.__logger.level
61 @property
62 def parent(self):
63 """
64 Get the logger parent.
66 Returns
67 -------
68 Logger
69 The logger parent.
70 """
71 return self.__logger.parent
73 @property
74 def propagate(self) -> bool:
75 """
76 Get the logger propagate flag.
78 Returns
79 -------
80 bool
81 The logger propagate flag.
82 """
83 return self.__logger.propagate
85 @property
86 def handlers(self) -> list[logging.Handler]:
87 """
88 Get the logger handlers.
90 Returns
91 -------
92 list[logging.Handler]
93 List of logger handlers.
94 """
95 return self.__logger.handlers
97 @property
98 def filters(self):
99 """
100 Get the logger filters.
102 Returns
103 -------
104 list
105 The logger filters.
106 """
107 return self.__logger.filters
109 @property
110 def disabled(self) -> bool:
111 """
112 Get the logger disabled flag.
114 Returns
115 -------
116 bool
117 The logger disabled flag.
118 """
119 return self.__logger.disabled
121 @property
122 def root(self):
123 """
124 Get the root logger.
126 Returns
127 -------
128 Logger
129 The root logger.
130 """
131 return self.__logger.root
133 @property
134 def manager(self) -> logging.Manager:
135 """
136 Get the logger manager.
138 Returns
139 -------
140 logging.Manager
141 The logger manager.
142 """
143 return self.__logger.manager
145 @property
146 def _cache(self):
147 """
148 Get the logger cache.
150 Returns
151 -------
152 dict
153 The logger cache.
154 """
155 # pylint: disable=protected-access
156 return self.__logger._cache
158 def __manage_logging(self, log_level: int, msg, *args, **kwargs):
159 """
160 Manage logging with optional message status and exchange integration.
162 Parameters
163 ----------
164 log_level : int
165 The log level.
166 msg
167 The log message.
168 *args
169 Additional arguments for the log message.
170 **kwargs
171 Additional keyword arguments.
172 """
173 message_status: LynceusMessageStatus | None = kwargs.pop(CONFIG_LOG_MESSAGE_STATUS_KEY, None)
175 # Defines automatically message status for warning and error.
176 if not message_status \
177 and log_level >= LynceusClientLogExchangeWrapper.AUTOMATIC_MESSAGE_STATUS_LOG_LEVEL_THRESHOLD:
178 message_status = LynceusMessageStatus.ERROR if log_level == ERROR else LynceusMessageStatus.WARNING
180 if self._lynceus_exchange:
181 # Creates an **object** somehow like Log Record, which will be consumed by caller (like API).
182 # queue_message: LogRecord = self.__logger.makeRecord(self.__logger.name, log_level, None, None, msg, args, exc_info=None, func=None, extra=None, sinfo=None)
183 queue_message = {
184 'asctime': datetime.now(tz=timezone.utc).strftime('%Y/%m/%d %H:%M:%S'),
185 'created': time.time(),
186 'levelname': logging.getLevelName(log_level),
187 'name': self.__logger.name,
188 'message': msg,
189 'message_status': message_status
190 }
192 # Puts the message in Lynceus exchange.
193 self._lynceus_exchange.put_nowait(queue_message)
195 # Requests true logging anyway.
196 if message_status:
197 msg += f' [{CONFIG_LOG_MESSAGE_STATUS_KEY}={message_status}]'
199 self.__logger.log(log_level, msg, *args, **kwargs)
201 def trace(self, msg, *args, **kwargs):
202 """
203 Log a trace message.
205 Parameters
206 ----------
207 msg
208 The log message.
209 *args
210 Additional arguments for the log message.
211 **kwargs
212 Additional keyword arguments.
213 """
214 # Intercepts the call, to registers message in the lynceus exchange.
215 self.__manage_logging(LynceusSession.TRACE, msg, *args, **kwargs)
217 def debug(self, msg, *args, **kwargs):
218 """
219 Log a debug message.
221 Parameters
222 ----------
223 msg
224 The log message.
225 *args
226 Additional arguments for the log message.
227 **kwargs
228 Additional keyword arguments.
229 """
230 # Intercepts the call, to registers message in the lynceus exchange.
231 self.__manage_logging(DEBUG, msg, *args, **kwargs)
233 def info(self, msg, *args, **kwargs):
234 """
235 Log an info message.
237 Parameters
238 ----------
239 msg
240 The log message.
241 *args
242 Additional arguments for the log message.
243 **kwargs
244 Additional keyword arguments.
245 """
246 # Intercepts the call, to registers message in the lynceus exchange.
247 self.__manage_logging(INFO, msg, *args, **kwargs)
249 def warning(self, msg, *args, **kwargs):
250 """
251 Log a warning message.
253 Parameters
254 ----------
255 msg
256 The log message.
257 *args
258 Additional arguments for the log message.
259 **kwargs
260 Additional keyword arguments.
261 """
262 # Intercepts the call, to registers message in the lynceus exchange.
263 self.__manage_logging(WARNING, msg, *args, **kwargs)
265 def error(self, msg, *args, **kwargs):
266 """
267 Log an error message.
269 Parameters
270 ----------
271 msg
272 The log message.
273 *args
274 Additional arguments for the log message.
275 **kwargs
276 Additional keyword arguments.
277 """
278 # Intercepts the call, to registers message in the lynceus exchange.
279 self.__manage_logging(ERROR, msg, *args, **kwargs)
281 def critical(self, msg, *args, **kwargs):
282 """
283 Log a critical message.
285 Parameters
286 ----------
287 msg
288 The log message.
289 *args
290 Additional arguments for the log message.
291 **kwargs
292 Additional keyword arguments.
293 """
294 # Intercepts the call, to registers message in the lynceus exchange.
295 self.__manage_logging(CRITICAL, msg, *args, **kwargs)
298class LynceusClientClass(metaclass=ABCMeta):
299 """
300 Base class for Lynceus client implementations.
301 """
303 def __init__(self, lynceus_session: LynceusSession, logger_name: str, lynceus_exchange: LynceusExchange | None):
304 """
305 Initialize the Lynceus client.
307 Parameters
308 ----------
309 lynceus_session : LynceusSession
310 The Lynceus session.
311 logger_name : str
312 Name for the logger.
313 lynceus_exchange : LynceusExchange or None
314 Optional exchange for message queuing.
315 """
316 self._lynceus_session: LynceusSession = lynceus_session
317 self._lynceus_exchange = lynceus_exchange
318 self.__logger: Logger = self._lynceus_session.get_logger(logger_name)
320 # Declares the self._logger attribute with Lynceus wrapper (with/out lynceus_exchange).
321 self._logger: LynceusClientLogExchangeWrapper = LynceusClientLogExchangeWrapper(self.__logger, self._lynceus_exchange)