distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Log and LogManager — I2P logging system wrapping Python's logging module.
2
3Ported from net.i2p.util.Log and net.i2p.util.LogManager.
4"""
5
6import logging
7import threading
8import traceback
9from typing import Dict, List, Optional
10
11
12class Log:
13 """I2P logger with custom priority levels."""
14
15 DEBUG = 10
16 INFO = 20
17 WARN = 30
18 ERROR = 40
19 CRIT = 50
20
21 STR_DEBUG = "DEBUG"
22 STR_INFO = "INFO"
23 STR_WARN = "WARN"
24 STR_ERROR = "ERROR"
25 STR_CRIT = "CRIT"
26
27 _LEVEL_MAP = {
28 "DEBUG": DEBUG, "INFO": INFO, "WARN": WARN,
29 "ERROR": ERROR, "CRIT": CRIT, "CRITICAL": CRIT,
30 }
31 _STR_MAP = {
32 DEBUG: STR_DEBUG, INFO: STR_INFO, WARN: STR_WARN,
33 ERROR: STR_ERROR, CRIT: STR_CRIT,
34 }
35
36 # Map I2P levels to Python logging levels
37 _PY_LEVEL_MAP = {
38 DEBUG: logging.DEBUG,
39 INFO: logging.INFO,
40 WARN: logging.WARNING,
41 ERROR: logging.ERROR,
42 CRIT: logging.CRITICAL,
43 }
44
45 def __init__(self, name: str, manager: "LogManager") -> None:
46 self._name = name
47 self._manager = manager
48 self._min_priority = Log.DEBUG
49 self._logger = logging.getLogger(f"i2p.{name}")
50
51 @staticmethod
52 def get_level(level_str: str) -> int:
53 return Log._LEVEL_MAP.get(level_str.upper(), Log.DEBUG)
54
55 @staticmethod
56 def to_level_string(level: int) -> str:
57 return Log._STR_MAP.get(level, "DEBUG")
58
59 @property
60 def name(self) -> str:
61 return self._name
62
63 def get_minimum_priority(self) -> int:
64 return self._min_priority
65
66 def set_minimum_priority(self, priority: int) -> None:
67 self._min_priority = priority
68
69 def should_log(self, priority: int) -> bool:
70 return priority >= self._min_priority
71
72 def should_debug(self) -> bool:
73 return self.should_log(Log.DEBUG)
74
75 def should_info(self) -> bool:
76 return self.should_log(Log.INFO)
77
78 def should_warn(self) -> bool:
79 return self.should_log(Log.WARN)
80
81 def should_error(self) -> bool:
82 return self.should_log(Log.ERROR)
83
84 def log(self, priority: int, msg: str, exc: Optional[Exception] = None) -> None:
85 if not self.should_log(priority):
86 return
87 py_level = self._PY_LEVEL_MAP.get(priority, logging.DEBUG)
88 if exc:
89 self._logger.log(py_level, msg, exc_info=exc)
90 else:
91 self._logger.log(py_level, msg)
92
93 def log_always(self, priority: int, msg: str) -> None:
94 py_level = self._PY_LEVEL_MAP.get(priority, logging.DEBUG)
95 self._logger.log(py_level, msg)
96
97 def debug(self, msg: str, exc: Optional[Exception] = None) -> None:
98 self.log(Log.DEBUG, msg, exc)
99
100 def info(self, msg: str, exc: Optional[Exception] = None) -> None:
101 self.log(Log.INFO, msg, exc)
102
103 def warn(self, msg: str, exc: Optional[Exception] = None) -> None:
104 self.log(Log.WARN, msg, exc)
105
106 def error(self, msg: str, exc: Optional[Exception] = None) -> None:
107 self.log(Log.ERROR, msg, exc)
108
109
110class LogManager:
111 """Central log manager — creates and manages Log instances."""
112
113 def __init__(self) -> None:
114 self._logs: Dict[str, Log] = {}
115 self._lock = threading.Lock()
116 self._default_limit = Log.WARN
117
118 def get_log(self, name_or_cls) -> Log:
119 """Get or create a Log for the given name or class."""
120 if isinstance(name_or_cls, type):
121 name = name_or_cls.__qualname__
122 else:
123 name = str(name_or_cls)
124
125 with self._lock:
126 if name not in self._logs:
127 log = Log(name, self)
128 log.set_minimum_priority(self._default_limit)
129 self._logs[name] = log
130 return self._logs[name]
131
132 def get_logs(self) -> List[Log]:
133 with self._lock:
134 return list(self._logs.values())
135
136 def get_default_limit(self) -> int:
137 return self._default_limit
138
139 def set_default_limit(self, level: int) -> None:
140 self._default_limit = level
141 with self._lock:
142 for log in self._logs.values():
143 log.set_minimum_priority(level)
144
145 def shutdown(self) -> None:
146 with self._lock:
147 self._logs.clear()