Skip to content

NatsParamsStorage

faststream.nats.broker.logging.NatsParamsStorage #

NatsParamsStorage()

Bases: DefaultLoggerStorage

Source code in faststream/nats/broker/logging.py 
14
15
16
17
18
19
20
21
def __init__(self) -> None:
    super().__init__()

    self._max_queue_len = 0
    self._max_stream_len = 0
    self._max_subject_len = 4

    self.logger_log_level = logging.INFO

logger_log_level instance-attribute #

logger_log_level = INFO

set_level #

set_level(level)
Source code in faststream/nats/broker/logging.py 
23
24
def set_level(self, level: int) -> None:
    self.logger_log_level = level

register_subscriber #

register_subscriber(params)
Source code in faststream/nats/broker/logging.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def register_subscriber(self, params: dict[str, Any]) -> None:
    self._max_subject_len = max(
        (
            self._max_subject_len,
            len(params.get("subject", "")),
        ),
    )
    self._max_queue_len = max(
        (
            self._max_queue_len,
            len(params.get("queue", "")),
        ),
    )
    self._max_stream_len = max(
        (
            self._max_stream_len,
            len(params.get("stream", "")),
        ),
    )

get_logger #

get_logger(*, context)
Source code in faststream/nats/broker/logging.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def get_logger(self, *, context: "ContextRepo") -> "LoggerProto":
    # TODO: generate unique logger names to not share between brokers
    if not (lg := self._get_logger_ref()):
        message_id_ln = 10

        lg = get_broker_logger(
            name="nats",
            default_context={
                "subject": "",
                "stream": "",
                "queue": "",
            },
            message_id_ln=message_id_ln,
            fmt="".join((
                "%(asctime)s %(levelname)-8s - ",
                (
                    f"%(stream)-{self._max_stream_len}s | "
                    if self._max_stream_len
                    else ""
                ),
                (
                    f"%(queue)-{self._max_queue_len}s | "
                    if self._max_queue_len
                    else ""
                ),
                f"%(subject)-{self._max_subject_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