Coverage for ntnlog / ntn_logging.py: 100%
134 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-21 04:50 +0000
1#######################################################################
2#
3# My Logging Module
4#
5# Provides logging functionality with timestamping and caller info
6# Supports configurable logging and tracing options
7#
8#######################################################################
10import asyncio
11import os
12import sys
13import traceback
14import threading
15from datetime import datetime
16from .ntn_file_utils import FileUtilsError, file_verify_path
17from inspect import getframeinfo, stack
18from .ntn_levels import Level, LevelLike
20from .ntn_config import (
21 GLOBAL_LOGGING_ENABLED,
22 GLOBAL_LOG_TRACING_ENABLED,
23 GLOBAL_LOG_LEVEL,
24 GLOBAL_MAX_BYTES,
25 GLOBAL_BACKUP_COUNT,
26 GLOBAL_LOG_COLORS,
27)
29_ANSI_RESET = "\033[0m"
32# ---------------------------------------------------------------------------
33# Frame-filtering constants
34# ---------------------------------------------------------------------------
36# Prefixes of filenames/paths to always skip (logger internals, frozen frames)
37_IGNORED_PREFIXES = (
38 "<frozen",
39 "<string>",
40)
42# Basenames of framework/infra files to skip (blocklist fallback mode)
43_IGNORED_FILES = {
44 "spawn.py",
45 "process.py",
46 "_subprocess.py",
47 "server.py",
48 "runners.py",
49 "config.py",
50 "importer.py",
51 "__init__.py",
52 "cli.py",
53 "discover.py",
54 "ntn_logging.py", # skip this file itself
55 "ntn_utils.py",
56}
58# ---------------------------------------------------------------------------
59# Logger
60# ---------------------------------------------------------------------------
62class Logger:
63 """
64 Lightweight file logger with optional call-stack tracing.
66 Usage
67 -----
68 log = Logger()
69 log("Something happened")
70 log.log("Something happened", console_message="Hey!")
72 Parameters
73 ----------
74 log_dir : str
75 Directory (relative or absolute) where log files are written.
76 Defaults to ``./logs``.
77 project_dir : str | None
78 Root of your project source tree. When supplied, only frames whose
79 absolute path starts with this directory (and are not inside
80 ``site-packages``) are considered "your code". When ``None`` the
81 module falls back to the blocklist defined by ``_IGNORED_FILES`` /
82 ``_IGNORED_PREFIXES``.
83 name : str | None
84 Optional label for this logger instance. Appears as a bracket segment
85 in every log entry.
86 level : LevelLike | None
87 Minimum level threshold for console output. Accepts a ``Level`` member
88 or a case-insensitive string: ``"TRACE"``, ``"DEBUG"``, ``"INFO"``,
89 ``"WARN"``, ``"ERROR"``, ``"CRITICAL"``. Entries are always written to
90 the file; only console output is suppressed below this threshold. When
91 ``None``, ``GLOBAL_LOG_LEVEL`` is used.
92 colorize : bool
93 When ``True``, console output is wrapped in ANSI color codes taken
94 from *colors* (or ``GLOBAL_LOG_COLORS`` when *colors* is ``None``).
95 File output is never colorized.
96 colors : dict[int, str] | None
97 Per-instance color map keyed by ``int(Level)``. Merged on top of
98 ``GLOBAL_LOG_COLORS`` at construction time; missing entries fall back
99 to the global value.
101 Notes
102 -----
103 Log rotation is controlled globally via ``GLOBAL_MAX_BYTES`` and
104 ``GLOBAL_BACKUP_COUNT`` in ``ntn_config.py``.
105 """
107 DEFAULT_LOG_DIR = "logs"
109 _LEVEL_ALIASES: dict[str, str] = {"WARN": "WARNING"}
111 @staticmethod
112 def _parse_level(value: LevelLike | None) -> Level | None:
113 if value is None:
114 return None
115 if isinstance(value, Level):
116 return value
117 name = str(value).upper()
118 name = Logger._LEVEL_ALIASES.get(name, name)
119 try:
120 return Level[name]
121 except KeyError:
122 valid = [lvl.name for lvl in Level]
123 raise ValueError(f"Unknown log level {value!r}. Valid levels: {valid}")
125 def __init__(
126 self,
127 log_dir: str = DEFAULT_LOG_DIR,
128 project_dir: str | None = None,
129 name: str | None = None,
130 level: LevelLike | None = None,
131 colorize: bool = False,
132 colors: dict[int, str] | None = None,
133 ):
134 self._enable: bool = True
135 self._enable_log_tracing: bool = False
136 self._log_dir: str = log_dir
137 self._project_dir: str | None = (
138 os.path.abspath(project_dir) if project_dir is not None else None
139 )
140 self._name: str | None = name
141 self._level: Level | None = self._parse_level(level)
142 self._colorize: bool = colorize
143 self._colors: dict[int, str] = {**GLOBAL_LOG_COLORS, **(colors or {})}
144 self._lock = threading.Lock()
146 # ------------------------------------------------------------------
147 # Public interface
148 # ------------------------------------------------------------------
150 def __call__(
151 self,
152 message: str,
153 level: LevelLike = Level.INFO,
154 console_message: str | None = None,
155 ) -> None:
156 """
157 Write a log entry. Alias for :meth:`log`.
159 Parameters
160 ----------
161 message : str
162 Text to write to the log file.
163 level : LevelLike
164 Severity of this entry. Always written to file. Suppresses console
165 output when below the instance threshold. Accepts a ``Level`` enum
166 member or a case-insensitive string: ``"TRACE"`` (5),
167 ``"DEBUG"`` (10), ``"INFO"`` (20), ``"WARN"`` (30),
168 ``"ERROR"`` (40), ``"CRITICAL"`` (50).
169 console_message : str | None
170 When ``None`` (default), nothing is printed to stdout.
171 When ``""`` (empty string), prints the full formatted log entry to stdout.
172 When any other string, prints that string to stdout instead of *message*.
173 """
174 self.log(message, level=level, console_message=console_message)
176 def log(
177 self,
178 message: str,
179 level: LevelLike = Level.INFO,
180 console_message: str | None = None,
181 ) -> None:
182 """
183 Write a log entry.
185 Parameters
186 ----------
187 message : str
188 Text to write to the log file.
189 level : LevelLike
190 Severity of this entry. Always written to file. Suppresses console
191 output when below the instance threshold. Accepts a ``Level`` enum
192 member or a case-insensitive string: ``"TRACE"`` (5),
193 ``"DEBUG"`` (10), ``"INFO"`` (20), ``"WARN"`` (30),
194 ``"ERROR"`` (40), ``"CRITICAL"`` (50).
195 console_message : str | None
196 When ``None`` (default), nothing is printed to stdout.
197 When ``""`` (empty string), prints the full formatted log entry to stdout.
198 When any other string, prints that string to stdout instead of *message*.
199 """
200 if isinstance(level, str):
201 level = self._parse_level(level) or Level.INFO
203 if not GLOBAL_LOGGING_ENABLED or not self._enable:
204 return
206 now = datetime.now()
207 date = now.strftime("%Y-%m-%d")
208 time_str = now.strftime("%H:%M:%S")
209 timestamp = f"{date} {time_str}"
211 name_segment = f"[{self._name}]" if self._name else ""
212 level_segment = f"[{level.name}]"
213 logistic_data = f"[{timestamp}]{level_segment}{name_segment}[{self.get_caller_info()}]"
214 padded_space = "".rjust(len(logistic_data) + 1)
215 formatted_message = str(message).replace("\n", f"\n{padded_space}")
216 log_entry = f"{logistic_data} {formatted_message}\n"
218 self._write_to_file(log_entry, date, time_str)
220 if console_message is not None:
221 threshold = self._level if self._level is not None else GLOBAL_LOG_LEVEL
222 if level >= threshold:
223 text = log_entry.rstrip("\n") if console_message == "" else console_message
224 if self._colorize:
225 color = self._colors.get(int(level), "")
226 text = f"{color}{text}{_ANSI_RESET}"
227 print(text)
229 def enable_logging(self, enable_logging: bool) -> None:
230 self._enable = enable_logging
232 def enable_log_tracing(self, enable_log_tracing: bool) -> None:
233 self._enable_log_tracing = enable_log_tracing
235 def exception(
236 self,
237 message: str,
238 level: LevelLike = Level.ERROR,
239 console_message: str | None = None,
240 ) -> None:
241 """
242 Log the current exception with a traceback appended.
244 Parameters
245 ----------
246 message : str
247 Text prepended to the traceback in the log entry.
248 level : LevelLike
249 Severity of this entry. Always written to file. Suppresses console
250 output when below the instance threshold. Accepts a ``Level`` enum
251 member or a case-insensitive string: ``"TRACE"`` (5),
252 ``"DEBUG"`` (10), ``"INFO"`` (20), ``"WARN"`` (30),
253 ``"ERROR"`` (40), ``"CRITICAL"`` (50).
254 console_message : str | None
255 When ``None`` (default), nothing is printed to stdout.
256 When ``""`` (empty string), prints the full formatted log entry to stdout.
257 When any other string, prints that string to stdout instead of *message*.
258 """
259 if sys.exc_info()[0] is None:
260 tb_text = ""
261 else:
262 tb_text = "\n" + traceback.format_exc().rstrip()
263 self.log(
264 f"{message}{tb_text}",
265 level=level,
266 console_message=console_message,
267 )
269 async def alog(
270 self,
271 message: str,
272 level: LevelLike = Level.INFO,
273 console_message: str | None = None,
274 ) -> None:
275 """
276 Async version of :meth:`log`.
278 Parameters
279 ----------
280 message : str
281 Text to write to the log file.
282 level : LevelLike
283 Severity of this entry. Always written to file. Suppresses console
284 output when below the instance threshold. Accepts a ``Level`` enum
285 member or a case-insensitive string: ``"TRACE"`` (5),
286 ``"DEBUG"`` (10), ``"INFO"`` (20), ``"WARN"`` (30),
287 ``"ERROR"`` (40), ``"CRITICAL"`` (50).
288 console_message : str | None
289 When ``None`` (default), nothing is printed to stdout.
290 When ``""`` (empty string), prints the full formatted log entry to stdout.
291 When any other string, prints that string to stdout instead of *message*.
292 """
293 await asyncio.to_thread(
294 self.log, message, level, console_message
295 )
297 async def aexception(
298 self,
299 message: str,
300 level: LevelLike = Level.ERROR,
301 console_message: str | None = None,
302 ) -> None:
303 """
304 Async version of :meth:`exception`.
306 Parameters
307 ----------
308 message : str
309 Text prepended to the traceback in the log entry.
310 level : LevelLike
311 Severity of this entry. Always written to file. Suppresses console
312 output when below the instance threshold. Accepts a ``Level`` enum
313 member or a case-insensitive string: ``"TRACE"`` (5),
314 ``"DEBUG"`` (10), ``"INFO"`` (20), ``"WARN"`` (30),
315 ``"ERROR"`` (40), ``"CRITICAL"`` (50).
316 console_message : str | None
317 When ``None`` (default), nothing is printed to stdout.
318 When ``""`` (empty string), prints the full formatted log entry to stdout.
319 When any other string, prints that string to stdout instead of *message*.
320 """
321 # Capture traceback now — sys.exc_info() is thread-local and will be
322 # empty inside the worker thread spawned by asyncio.to_thread().
323 if sys.exc_info()[0] is None:
324 tb_text = ""
325 else:
326 tb_text = "\n" + traceback.format_exc().rstrip()
327 await asyncio.to_thread(
328 self.log,
329 f"{message}{tb_text}",
330 level,
331 console_message,
332 )
334 # ------------------------------------------------------------------
335 # Caller info
336 # ------------------------------------------------------------------
338 def get_caller_info(self) -> str:
339 tracing_active = GLOBAL_LOG_TRACING_ENABLED and self._enable_log_tracing
340 frames = self._collect_project_frames()
342 if not frames:
343 return "unknown:0"
345 if not tracing_active:
346 return frames[0]
348 return ">".join(frames)
350 # ------------------------------------------------------------------
351 # Private helpers
352 # ------------------------------------------------------------------
354 def _is_project_frame(self, filename: str) -> bool:
355 """
356 Return True when *filename* belongs to project code.
358 Allowlist mode (project_dir set):
359 Frame must live inside the project directory and not inside
360 a ``site-packages`` folder.
362 Blocklist mode (project_dir is None):
363 Frame must not match any entry in ``_IGNORED_FILES`` or
364 ``_IGNORED_PREFIXES``.
365 """
366 # Always skip frozen / string pseudo-files regardless of mode
367 if any(filename.startswith(prefix) for prefix in _IGNORED_PREFIXES):
368 return False
370 if self._project_dir is not None:
371 abs_filename = os.path.abspath(filename)
372 return (
373 abs_filename.startswith(self._project_dir)
374 and "site-packages" not in abs_filename
375 )
377 # Blocklist fallback
378 return os.path.basename(filename) not in _IGNORED_FILES
380 def _collect_project_frames(self) -> list[str]:
381 """Walk the call stack and return frame strings for project code only."""
382 frames: list[str] = []
383 for frame_record in stack():
384 try:
385 info = getframeinfo(frame_record[0])
386 if self._is_project_frame(info.filename):
387 frames.append(f"{os.path.basename(info.filename)}:{info.lineno}")
388 except (IndexError, TypeError):
389 # IndexError: stack shorter than expected
390 # TypeError: frame_record is not subscriptable (e.g. test mocks using None)
391 continue
392 return frames
394 def _write_to_file(self, log_entry: str, date: str, time_str: str) -> None:
395 """Create the log directory/file if needed, then append *log_entry*."""
396 log_dir_path = os.path.join("./", self._log_dir)
398 # Ensure the log directory exists
399 file_path = file_verify_path("./", self._log_dir)
400 if file_path == FileUtilsError.NOT_A_DIRECTORY.value.format(directory=self._log_dir):
401 os.makedirs(log_dir_path, exist_ok=True)
402 file_path = file_verify_path("./", self._log_dir) # re-verify after creation
404 # Bail out if the path is still an error (e.g. outside working dir)
405 if not isinstance(file_path, str) or file_path.startswith("Error"):
406 return
408 file_name = os.path.join(file_path, f"{date}_logging.txt")
410 max_bytes = GLOBAL_MAX_BYTES
411 backup_count = GLOBAL_BACKUP_COUNT
413 with self._lock:
414 if os.path.exists(file_name) and os.path.getsize(file_name) >= max_bytes:
415 if backup_count == 0:
416 os.remove(file_name)
417 else:
418 oldest = f"{file_name}.{backup_count}"
419 if os.path.exists(oldest):
420 os.remove(oldest)
421 for i in range(backup_count - 1, 0, -1):
422 src = f"{file_name}.{i}"
423 dst = f"{file_name}.{i + 1}"
424 if os.path.exists(src):
425 os.rename(src, dst)
426 os.rename(file_name, f"{file_name}.1")
428 with open(file_name, "a") as log_file:
429 # Write a header only on the very first entry (empty file)
430 if log_file.tell() == 0:
431 log_file.write(f"Log file created on {date} at {time_str}\n")
432 log_file.write(log_entry)