1#!/usr/bin/env python3
2"""
3
4"""
5# mypy: disable-error-code="attr-defined"
6# Imports:
7from __future__ import annotations
8
9# ##-- stdlib imports
10import datetime
11import functools as ftz
12import itertools as itz
13import logging as logmod
14import logging.handlers as l_handlers
15import os
16import pathlib as pl
17import re
18import time
19import types
20import weakref
21from sys import stderr, stdout
22from uuid import UUID, uuid1
23
24# ##-- end stdlib imports
25
26# ##-- 1st party imports
27from jgdv import Mixin, Proto
28from jgdv._abstract.protocols.general import Buildable_p
29from jgdv._abstract.protocols.pydantic import ProtocolModelMeta
30from jgdv.structs.chainguard import ChainGuard
31
32# ##-- end 1st party imports
33
34from . import _interface as API # noqa: N812
35from .filter import BlacklistFilter, WhitelistFilter
36from .format import ColourFormatter, StripColourFormatter
37
38# ##-- types
39# isort: off
40import abc
41import collections.abc
42from typing import TYPE_CHECKING, cast, assert_type, assert_never
43from typing import Generic, NewType, Any
44# Protocols:
45from typing import Protocol, runtime_checkable
46# Typing Decorators:
47from typing import no_type_check, final, override, overload
48from pydantic import BaseModel, Field, model_validator, field_validator, ValidationError
49if TYPE_CHECKING:
50 import enum
51 from jgdv import Maybe, RxStr
52 from typing import Final
53 from typing import ClassVar, Any, LiteralString
54 from typing import Never, Self, Literal
55 from typing import TypeGuard
56 from collections.abc import Iterable, Iterator, Callable, Generator
57 from collections.abc import Sequence, Mapping, MutableMapping, Hashable
58 from ._interface import Handler, Formatter
59
60##--|
61from jgdv import Maybe
62# isort: on
63# ##-- end types
64
65##-- logging
66logging = logmod.getLogger(__name__)
67##-- end logging
68
69env : dict = cast("dict", os.environ)
70IS_PRE_COMMIT : Final[bool] = "PRE_COMMIT" in env
71DEFAULT_FORMAT : Final[str] = "{levelname:<8} : {message}"
72DEFAULT_FILE_FORMAT : Final[str] = "%Y-%m-%d::%H:%M.log"
73DEFAULT_STYLE : Final[str] = "{"
74##--|
[docs]
75class HandlerBuilder_m:
76 """
77 Loggerspec Mixin for building handlers
78 """
79
[docs]
80 def _build_streamhandler(self) -> Handler:
81 return logmod.StreamHandler(stdout)
82
[docs]
83 def _build_errorhandler(self) -> Handler:
84 return logmod.StreamHandler(stderr)
85
[docs]
86 def _build_filehandler(self, path:pl.Path) -> Handler:
87 return logmod.FileHandler(path, mode='w')
88
[docs]
89 def _build_rotatinghandler(self, path:pl.Path) -> Handler:
90 handler = l_handlers.RotatingFileHandler(path, backupCount=API.MAX_FILES)
91 handler.doRollover()
92 return handler
93
110
[docs]
111 def _build_filters(self) -> list[Callable]:
112 filters : list[Callable] = []
113 if bool(self.allow):
114 filters.append(WhitelistFilter(self.allow))
115 if bool(self.filter):
116 filters.append(BlacklistFilter(self.filter))
117
118 return filters
119
[docs]
120 def _discriminate_handler(self, target:Maybe[str|pl.Path]) -> tuple[Maybe[Handler], Maybe[Formatter]]:
121 handler, formatter = None, None
122
123 match target:
124 case "pass" | None:
125 return None, None
126 case "file":
127 log_file_path = self.logfile()
128 handler = self._build_filehandler(log_file_path)
129 case "rotate":
130 log_file_path = self.logfile()
131 handler = self._build_rotatinghandler(log_file_path)
132 case "stdout":
133 handler = self._build_streamhandler()
134 case "stderr":
135 handler = self._build_errorhandler()
136 case _:
137 msg = "Unknown logger spec target"
138 raise ValueError(msg, target)
139
140 formatter = self._build_formatter(handler)
141
142 assert(handler is not None)
143 assert(formatter is not None)
144 return handler, formatter
145
146##--|
147
[docs]
148@Proto(Buildable_p)
149class LoggerSpec(HandlerBuilder_m, BaseModel, metaclass=ProtocolModelMeta):
150 """
151 A Spec for toml defined logging control.
152 Allows user to name a logger, set its level, format,
153 filters, colour, and what (cli arg) verbosity it activates on,
154 and what file it logs to.
155
156 When 'apply' is called, it gets the logger,
157 and sets any relevant settings on it.
158 """
159 ##--| classvars
160 RootName : ClassVar[str] = "root"
161 levels : ClassVar[type[enum.IntEnum]] = API.LogLevel_e
162 ##--| main
163 name : str
164 disabled : bool = False
165 base : Maybe[str] = None
166 level : str|int = logmod.WARNING
167 format : str = DEFAULT_FORMAT
168 filter : list[str] = []
169 allow : list[str] = []
170 colour : bool|str = False
171 verbosity : int = 0
172 target : list[str|pl.Path] = [] # stdout | stderr | file
173 filename_fmt : str = DEFAULT_FILE_FORMAT
174 propagate : bool = False
175 clear_handlers : bool = False
176 style : str = DEFAULT_STYLE
177 nested : list[LoggerSpec] = []
178 prefix : Maybe[str] = None
179 ##--| internal
180 _logger : Maybe[API.Logger] = None
181 _applied : bool = False
182
[docs]
183 @staticmethod
184 def build(data:bool|list|dict, **kwargs:Any) -> LoggerSpec: # noqa: ANN401, FBT001
185 """
186 Build a single spec, or multiple logger specs targeting the same logger
187 """
188 match data:
189 case LoggerSpec():
190 return data
191 case False:
192 return LoggerSpec(disabled=True, **kwargs)
193 case True:
194 return LoggerSpec(**kwargs)
195 case list():
196 nested = []
197 for x in data:
198 nested.append(LoggerSpec.build(x, **kwargs))
199 return LoggerSpec(nested=nested, **kwargs)
200 case ChainGuard():
201 as_dict = dict(data)
202 as_dict.update(kwargs)
203 return LoggerSpec.model_validate(as_dict)
204 case dict():
205 as_dict = data.copy()
206 as_dict.update(kwargs)
207 return LoggerSpec.model_validate(as_dict)
208 case _:
209 msg = "Unknown data for logger spec"
210 raise TypeError(msg, data)
211
212 ##--| Validators
213
[docs]
214 @field_validator("level")
215 def _validate_level(cls, val:str|int) -> int: # noqa: N805
216 match val:
217 case str() if (lvl:=logmod.getLevelNamesMapping().get(val, None)) is not None:
218 return lvl
219 case str() if val in LoggerSpec.levels.__members__:
220 return LoggerSpec.levels[val]
221 case int():
222 return val
223 case _:
224 raise ValueError(val)
225
229
[docs]
230 @field_validator("target", mode="before")
231 def _validate_target(cls, val:list|str|pl.Path) -> list[str|pl.Path]: # noqa: N805
232 match val:
233 case [*xs] if all(x in API.TARGETS for x in xs):
234 return val
235 case str() if val in API.TARGETS:
236 return [val]
237 case pl.Path():
238 return [val]
239 case None:
240 return ["stdout"]
241 case _:
242 msg = "Unknown target value for LoggerSpec"
243 raise ValueError(msg, val)
244
[docs]
245 @field_validator("style")
246 def _validate_style(cls, val:str) -> str: # noqa: N805
247 match val:
248 case "%" | "{" | "$":
249 return val
250 case _:
251 msg = "API.Logger Style Needs to be in [{,%,$]"
252 raise ValueError(msg, val)
253
254 ##--| methods
255
[docs]
256 @ftz.cached_property
257 def fullname(self) -> str:
258 if self.base is None:
259 return self.name
260 return f"{self.base}.{self.name}"
261
[docs]
262 def apply(self, *, onto:Maybe[API.Logger]=None) -> API.Logger: # noqa: PLR0912, PLR0915
263 """ Apply this spec (and nested specs) to the relevant logger """
264 logger : logmod.Logger
265 match onto:
266 case logmod.Logger() if self._applied:
267 msg = "Tried to apply logger when spec already has a logger"
268 raise ValueError(msg, self.fullname)
269 case logmod.Logger():
270 logger = onto
271 case None if self._applied:
272 # already set up, just return it
273 return self.get()
274 case None:
275 # not set up, get it and set it
276 logger = self.get()
277
278 handler_pairs : list[tuple[Maybe[Handler], Maybe[Formatter]]] = []
279 logger.propagate = self.propagate
280 logger.setLevel(logmod._nameToLevel.get("NOTSET", 0))
281 if self.disabled:
282 logger.disabled = True
283 return logger
284
285 match self.prefix:
286 case str() if hasattr(logger, "set_prefixes"):
287 logger.set_prefixes(self.prefix)
288 case _:
289 pass
290
291 match self.colour:
292 case str():
293 logger.set_colour(self.colour)
294 case _:
295 pass
296
297 match self.target:
298 case _ if bool(self.nested):
299 for subspec in self.nested:
300 subspec.apply(onto=logger)
301 else:
302 return logger
303 case []:
304 handler_pairs.append(self._discriminate_handler(None))
305 case [*xs]:
306 handler_pairs += [self._discriminate_handler(x) for x in xs]
307 case _:
308 msg = "Unknown target value for LoggerSpec"
309 raise ValueError(msg, self.target)
310
311 log_filters = self._build_filters()
312 for pair in handler_pairs:
313 match pair:
314 case None, _:
315 pass
316 case hand, None:
317 hand.setLevel(self.level)
318 for fltr in log_filters:
319 hand.addFilter(fltr)
320 else:
321 logger.addHandler(hand)
322 case hand, fmt:
323 hand.setLevel(self.level)
324 hand.setFormatter(fmt)
325 for fltr in log_filters:
326 hand.addFilter(fltr)
327 else:
328 logger.addHandler(hand)
329 case _:
330 pass
331 else:
332 if not bool(logger.handlers):
333 logger.setLevel(self.level)
334 logger.propagate = True
335
336 self._applied = True
337 self._logger = logger
338 return logger
339
[docs]
340 def get(self) -> API.Logger:
341 """ Get the logger this spec controls """
342 if self._logger is None:
343 self._logger = logmod.getLogger(self.fullname)
344
345 return self._logger
346
[docs]
347 def clear(self) -> None:
348 """ Clear the handlers for the logger referenced """
349 logger = self.get()
350 handlers = logger.handlers[:]
351 for h in handlers:
352 logger.removeHandler(h)
353
354 self._logger = None
355
[docs]
356 def logfile(self) -> pl.Path:
357 log_dir = pl.Path(".temp/logs")
358 if not log_dir.exists():
359 log_dir = pl.Path()
360
361 filename = datetime.datetime.now().strftime(self.filename_fmt) # noqa: DTZ005
362 return log_dir / filename
363
[docs]
364 def set_level(self, level:int|str) -> None:
365 match level:
366 case str():
367 level = logmod._nameToLevel.get(level, 0)
368 case int():
369 pass
370 logger = self.get()
371 logger.setLevel(level)
372 for handler in logger.handlers:
373 handler.setLevel(level)