Coverage for muutils/logger/headerfuncs.py: 95%

19 statements  

« prev     ^ index     » next       coverage.py v7.6.1, created at 2025-04-04 03:33 -0600

1from __future__ import annotations 

2 

3import json 

4from typing import Any, Mapping, Protocol 

5 

6from muutils.json_serialize import json_serialize 

7 

8# takes message, level, other data, and outputs message with appropriate header 

9# HeaderFunction = Callable[[str, int, Any], str] 

10 

11 

12class HeaderFunction(Protocol): 

13 def __call__(self, msg: Any, lvl: int, **kwargs) -> str: ... 

14 

15 

16def md_header_function( 

17 msg: Any, 

18 lvl: int, 

19 stream: str | None = None, 

20 indent_lvl: str = " ", 

21 extra_indent: str = "", 

22 **kwargs, 

23) -> str: 

24 """standard header function. will output 

25 

26 - `# {msg}` 

27 

28 for levels in [0, 9] 

29 

30 - `## {msg}` 

31 

32 for levels in [10, 19], and so on 

33 

34 - `[{stream}] # {msg}` 

35 

36 for a non-`None` stream, with level headers as before 

37 

38 - `!WARNING! [{stream}] {msg}` 

39 

40 for level in [-9, -1] 

41 

42 - `!!WARNING!! [{stream}] {msg}` 

43 

44 for level in [-19, -10] and so on 

45 

46 """ 

47 stream_prefix: str = "" 

48 if stream is not None: 

49 stream_prefix = f"[{stream}] " 

50 

51 lvl_div_10: int = lvl // 10 

52 

53 msg_processed: str 

54 if isinstance(msg, Mapping): 

55 msg_processed = ", ".join([f"{k}: {json_serialize(v)}" for k, v in msg.items()]) 

56 else: 

57 msg_processed = json.dumps(json_serialize(msg)) 

58 

59 if lvl >= 0: 

60 return f"{extra_indent}{indent_lvl * (lvl_div_10 - 1)}{stream_prefix}#{'#' * lvl_div_10 if lvl else ''} {msg_processed}" 

61 else: 

62 exclamation_pts: str = "!" * (abs(lvl) // 10) 

63 return f"{extra_indent}{exclamation_pts}WARNING{exclamation_pts} {stream_prefix} {msg_processed}" 

64 

65 

66HEADER_FUNCTIONS: dict[str, HeaderFunction] = { 

67 "md": md_header_function, 

68}