Bases: DefaultLoggerStorage
Source code in faststream/mqtt/broker/logging.py
| def __init__(self) -> None:
super().__init__()
self._max_topic_len = 4
self._max_shared_len = 0
self.logger_log_level = logging.INFO
|
instance-attribute
logger_log_level = logging.INFO
set_level(level: int) -> None
Source code in faststream/mqtt/broker/logging.py
| def set_level(self, level: int) -> None:
self.logger_log_level = level
|
register_subscriber(params: dict[str, Any]) -> None
Source code in faststream/mqtt/broker/logging.py
29
30
31
32
33
34
35
36
37 | def register_subscriber(self, params: dict[str, Any]) -> None:
self._max_topic_len = max(
self._max_topic_len,
len(params.get("topic", "")),
)
self._max_shared_len = max(
self._max_shared_len,
len(params.get("shared", "") or ""),
)
|
Source code in faststream/mqtt/broker/logging.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66 | def get_logger(self, *, context: "ContextRepo") -> "LoggerProto":
if not (lg := self._get_logger_ref()):
message_id_ln = 10
lg = get_broker_logger(
name="mqtt",
default_context={
"topic": "",
"shared": "",
},
message_id_ln=message_id_ln,
fmt="".join((
"%(asctime)s %(levelname)-8s - ",
(
f"%(shared)-{self._max_shared_len}s | "
if self._max_shared_len
else ""
),
f"%(topic)-{self._max_topic_len}s | ",
f"%(message_id)-{message_id_ln}s - ",
"%(message)s",
)),
context=context,
log_level=self.logger_log_level,
)
self._logger_ref.add(lg)
return lg
|