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
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
| 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
| 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
|