Skip to content

logging

scrapli.logging

ScrapliFileHandler

Bases: FileHandler_

Source code in scrapli/logging.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class ScrapliFileHandler(FileHandler_):
    def __init__(
        self,
        filename: str,
        mode: str = "a",
        encoding: Optional[str] = None,
        delay: bool = False,
    ) -> None:
        """
        Handle "buffering" log read messages for logging.FileHandler

        Args:
            filename: name of file to create
            mode: file mode
            encoding: encoding to use for file
            delay: actually not sure what this is for :)

        Returns:
            None

        Raises:
            N/A

        """
        super().__init__(
            filename=filename,
            mode=mode,
            encoding=encoding,
            delay=delay,
        )
        self._record_buf: Optional[LogRecord_] = None
        self._record_msg_buf: bytes = b""
        self._read_msg_prefix = "read: "
        self._read_msg_prefix_len = len(self._read_msg_prefix)

    def emit_buffered(self) -> None:
        """
        Emit a buffered read message to the FileHandler

        Args:
            N/A

        Returns:
            None

        Raises:
            ScrapliException: should never be raised!

        """
        if not self._record_buf:
            raise ScrapliException(
                "something unexpected happened in the ScrapliFileHandler log handler"
            )

        self._record_buf.msg = f"read : {self._record_msg_buf!r}"
        super().emit(record=self._record_buf)
        self._record_buf = None
        self._record_msg_buf = b""

    def emit(self, record: LogRecord_) -> None:
        """
        Override standard library FileHandler.emit to "buffer" subsequent read messages

        Args:
            record: log record to check

        Returns:
            None

        Raises:
            N/A

        """
        if not record.msg.startswith(self._read_msg_prefix):
            # everytime we get a message *not* starting with "read: " we check to see if there is
            # any buffered message ready to send, if so send it. otherwise, treat the message
            # normally by super'ing to the "normal" handler
            if self._record_buf:
                self.emit_buffered()

            super().emit(record=record)
            return

        if self._record_buf is None:
            # no message in the buffer, set the current record to the _record_buf
            self._record_buf = record
            # get the payload of the message after "read: " and re-convert it to bytes
            self._record_msg_buf = literal_eval(record.msg[self._read_msg_prefix_len :])  # noqa
            return

        # if we get here we know we are getting subsequent read messages we want to buffer -- the
        # log record data will all be the same, its just the payload that will be new, so add that
        # current payload to the _record_msg_buf buffer
        self._record_msg_buf += literal_eval(record.msg[self._read_msg_prefix_len :])  # noqa

__init__(filename: str, mode: str = 'a', encoding: Optional[str] = None, delay: bool = False) -> None

Handle "buffering" log read messages for logging.FileHandler

Parameters:

Name Type Description Default
filename str

name of file to create

 required
mode str

file mode

 'a'
encoding Optional[str]

encoding to use for file

 None
delay bool

actually not sure what this is for :)

 False

Returns:

Type Description
None

None
Source code in scrapli/logging.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def __init__(
    self,
    filename: str,
    mode: str = "a",
    encoding: Optional[str] = None,
    delay: bool = False,
) -> None:
    """
    Handle "buffering" log read messages for logging.FileHandler

    Args:
        filename: name of file to create
        mode: file mode
        encoding: encoding to use for file
        delay: actually not sure what this is for :)

    Returns:
        None

    Raises:
        N/A

    """
    super().__init__(
        filename=filename,
        mode=mode,
        encoding=encoding,
        delay=delay,
    )
    self._record_buf: Optional[LogRecord_] = None
    self._record_msg_buf: bytes = b""
    self._read_msg_prefix = "read: "
    self._read_msg_prefix_len = len(self._read_msg_prefix)

emit(record: LogRecord_) -> None

Override standard library FileHandler.emit to "buffer" subsequent read messages

Parameters:

Name Type Description Default
record LogRecord_

log record to check

 required

Returns:

Type Description
None

None
Source code in scrapli/logging.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def emit(self, record: LogRecord_) -> None:
    """
    Override standard library FileHandler.emit to "buffer" subsequent read messages

    Args:
        record: log record to check

    Returns:
        None

    Raises:
        N/A

    """
    if not record.msg.startswith(self._read_msg_prefix):
        # everytime we get a message *not* starting with "read: " we check to see if there is
        # any buffered message ready to send, if so send it. otherwise, treat the message
        # normally by super'ing to the "normal" handler
        if self._record_buf:
            self.emit_buffered()

        super().emit(record=record)
        return

    if self._record_buf is None:
        # no message in the buffer, set the current record to the _record_buf
        self._record_buf = record
        # get the payload of the message after "read: " and re-convert it to bytes
        self._record_msg_buf = literal_eval(record.msg[self._read_msg_prefix_len :])  # noqa
        return

    # if we get here we know we are getting subsequent read messages we want to buffer -- the
    # log record data will all be the same, its just the payload that will be new, so add that
    # current payload to the _record_msg_buf buffer
    self._record_msg_buf += literal_eval(record.msg[self._read_msg_prefix_len :])  # noqa

emit_buffered() -> None

Emit a buffered read message to the FileHandler

Returns:

Type Description
None

None

Raises:

Type Description
ScrapliException

should never be raised!
Source code in scrapli/logging.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def emit_buffered(self) -> None:
    """
    Emit a buffered read message to the FileHandler

    Args:
        N/A

    Returns:
        None

    Raises:
        ScrapliException: should never be raised!

    """
    if not self._record_buf:
        raise ScrapliException(
            "something unexpected happened in the ScrapliFileHandler log handler"
        )

    self._record_buf.msg = f"read : {self._record_msg_buf!r}"
    super().emit(record=self._record_buf)
    self._record_buf = None
    self._record_msg_buf = b""

ScrapliFormatter

Bases: Formatter_

Source code in scrapli/logging.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 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
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class ScrapliFormatter(Formatter_):
    def __init__(self, log_header: bool = True, caller_info: bool = False) -> None:
        """
        Scrapli's opinionated custom log formatter class

        Only applied/used when explicitly requested by the user, otherwise we leave logging up to
        the user as any library should!

        Args:
            log_header: add the "header" row to logging output (or not)
            caller_info: add caller (module/package/line) info to log output

        Returns:
            None

        Raises:
            N/A

        """
        log_format = "{message_id:<5} | {asctime} | {levelname:<8} | {target: <25} | {message}"
        if caller_info:
            log_format = (
                "{message_id:<5} | {asctime} | {levelname:<8} | {target: <25} | "
                "{module:<20} | {funcName:<20} | {lineno:<5} | {message}"
            )

        super().__init__(fmt=log_format, style="{")

        self.log_header = log_header
        self.caller_info = caller_info
        self.message_id = 1

        self.header_record = ScrapliLogRecord(
            name="header",
            level=0,
            pathname="",
            lineno=0,
            msg="MESSAGE",
            args=(),
            exc_info=None,
        )
        self.header_record.message_id = 0
        self.header_record.asctime = "TIMESTAMP".ljust(23, " ")
        self.header_record.levelname = "LEVEL"
        self.header_record.uid = "(UID:)"
        self.header_record.host = "HOST"
        self.header_record.port = "PORT"
        self.header_record.module = "MODULE"
        self.header_record.funcName = "FUNCNAME"
        self.header_record.lineno = 0
        self.header_record.message = "MESSAGE"

    def formatMessage(self, record: LogRecord_) -> str:
        """
        Override standard library logging Formatter.formatMessage

        Args:
            record: LogRecord to format

        Returns:
            str: log string to emit

        Raises:
            N/A

        """
        record = cast(ScrapliLogRecord, record)

        record.message_id = self.message_id

        if not hasattr(record, "host"):
            # if no host/port set, assign to the record so formatting does not fail
            record.host = ""
            record.port = ""
            _host_port = ""
        else:
            _host_port = f"{record.host}:{record.port}"

        _uid = "" if not hasattr(record, "uid") else f"{record.uid}:"
        # maybe this name changes... but a uid in the event you have multiple connections to a
        # single host... w/ this you can assign the uid so you know which is which
        record.target = f"{_uid}{_host_port}"
        # add colon to the uid so the log messages are pretty
        record.target = (
            record.target[:25] if len(record.target) <= 25 else f"{record.target[:22]}..."
        )

        if self.caller_info:
            record.module = (
                record.module[:20] if len(record.module) <= 20 else f"{record.module[:17]}..."
            )
            record.funcName = (
                record.funcName[:20] if len(record.funcName) <= 20 else f"{record.funcName[:17]}..."
            )

        message = self._style.format(record)

        if self.message_id == 1 and self.log_header:
            # ignoring type for these fields so we can put "pretty" data into the log "header" row
            self.header_record.message_id = "ID"  # type: ignore
            self.header_record.lineno = "LINE"  # type: ignore
            self.header_record.target = "(UID:)HOST:PORT".ljust(len(record.target))
            header_message = self._style.format(self.header_record)
            message = header_message + "\n" + message

        self.message_id += 1

        return message

__init__(log_header: bool = True, caller_info: bool = False) -> None

Scrapli's opinionated custom log formatter class

Only applied/used when explicitly requested by the user, otherwise we leave logging up to the user as any library should!

Parameters:

Name Type Description Default
log_header bool

add the "header" row to logging output (or not)

 True
caller_info bool

add caller (module/package/line) info to log output

 False

Returns:

Type Description
None

None
Source code in scrapli/logging.py 
29
30
31
32
33
34
35
36
37
38
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
67
68
69
70
71
72
73
74
75
76
77
78
def __init__(self, log_header: bool = True, caller_info: bool = False) -> None:
    """
    Scrapli's opinionated custom log formatter class

    Only applied/used when explicitly requested by the user, otherwise we leave logging up to
    the user as any library should!

    Args:
        log_header: add the "header" row to logging output (or not)
        caller_info: add caller (module/package/line) info to log output

    Returns:
        None

    Raises:
        N/A

    """
    log_format = "{message_id:<5} | {asctime} | {levelname:<8} | {target: <25} | {message}"
    if caller_info:
        log_format = (
            "{message_id:<5} | {asctime} | {levelname:<8} | {target: <25} | "
            "{module:<20} | {funcName:<20} | {lineno:<5} | {message}"
        )

    super().__init__(fmt=log_format, style="{")

    self.log_header = log_header
    self.caller_info = caller_info
    self.message_id = 1

    self.header_record = ScrapliLogRecord(
        name="header",
        level=0,
        pathname="",
        lineno=0,
        msg="MESSAGE",
        args=(),
        exc_info=None,
    )
    self.header_record.message_id = 0
    self.header_record.asctime = "TIMESTAMP".ljust(23, " ")
    self.header_record.levelname = "LEVEL"
    self.header_record.uid = "(UID:)"
    self.header_record.host = "HOST"
    self.header_record.port = "PORT"
    self.header_record.module = "MODULE"
    self.header_record.funcName = "FUNCNAME"
    self.header_record.lineno = 0
    self.header_record.message = "MESSAGE"

formatMessage(record: LogRecord_) -> str

Override standard library logging Formatter.formatMessage

Parameters:

Name Type Description Default
record LogRecord_

LogRecord to format

 required

Returns:

Name Type Description
str str

log string to emit
Source code in scrapli/logging.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def formatMessage(self, record: LogRecord_) -> str:
    """
    Override standard library logging Formatter.formatMessage

    Args:
        record: LogRecord to format

    Returns:
        str: log string to emit

    Raises:
        N/A

    """
    record = cast(ScrapliLogRecord, record)

    record.message_id = self.message_id

    if not hasattr(record, "host"):
        # if no host/port set, assign to the record so formatting does not fail
        record.host = ""
        record.port = ""
        _host_port = ""
    else:
        _host_port = f"{record.host}:{record.port}"

    _uid = "" if not hasattr(record, "uid") else f"{record.uid}:"
    # maybe this name changes... but a uid in the event you have multiple connections to a
    # single host... w/ this you can assign the uid so you know which is which
    record.target = f"{_uid}{_host_port}"
    # add colon to the uid so the log messages are pretty
    record.target = (
        record.target[:25] if len(record.target) <= 25 else f"{record.target[:22]}..."
    )

    if self.caller_info:
        record.module = (
            record.module[:20] if len(record.module) <= 20 else f"{record.module[:17]}..."
        )
        record.funcName = (
            record.funcName[:20] if len(record.funcName) <= 20 else f"{record.funcName[:17]}..."
        )

    message = self._style.format(record)

    if self.message_id == 1 and self.log_header:
        # ignoring type for these fields so we can put "pretty" data into the log "header" row
        self.header_record.message_id = "ID"  # type: ignore
        self.header_record.lineno = "LINE"  # type: ignore
        self.header_record.target = "(UID:)HOST:PORT".ljust(len(record.target))
        header_message = self._style.format(self.header_record)
        message = header_message + "\n" + message

    self.message_id += 1

    return message

enable_basic_logging(file: Union[str, bool] = False, level: str = 'info', caller_info: bool = False, buffer_log: bool = True, mode: str = 'write') -> None

Enable opinionated logging for scrapli

Parameters:

Name Type Description Default
file Union[str, bool]

True to output to default log path ("scrapli.log"), otherwise string path to write log file to

 False
level str

string name of logging level to use, i.e. "info", "debug", etc.

 'info'
caller_info bool

add info about module/function/line in the log entry

 False
buffer_log bool

buffer log read outputs

 True
mode str

string of "write" or "append"

 'write'

Returns:

Type Description
None

None

Raises:

Type Description
ScrapliException

if invalid mode is passed
Source code in scrapli/logging.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def enable_basic_logging(
    file: Union[str, bool] = False,
    level: str = "info",
    caller_info: bool = False,
    buffer_log: bool = True,
    mode: str = "write",
) -> None:
    """
    Enable opinionated logging for scrapli

    Args:
        file: True to output to default log path ("scrapli.log"), otherwise string path to write log
            file to
        level: string name of logging level to use, i.e. "info", "debug", etc.
        caller_info: add info about module/function/line in the log entry
        buffer_log: buffer log read outputs
        mode: string of "write" or "append"

    Returns:
        None

    Raises:
        ScrapliException: if invalid mode is passed

    """
    logger.propagate = False
    logger.setLevel(level=level.upper())

    scrapli_formatter = ScrapliFormatter(caller_info=caller_info)

    if mode.lower() not in (
        "write",
        "append",
    ):
        raise ScrapliException("logging file 'mode' must be 'write' or 'append'!")
    file_mode = "a" if mode.lower() == "append" else "w"

    if file:
        filename = "scrapli.log" if isinstance(file, bool) else file
        if not buffer_log:
            fh = FileHandler_(filename=filename, mode=file_mode)
        else:
            fh = ScrapliFileHandler(filename=filename, mode=file_mode)

        fh.setFormatter(scrapli_formatter)

        logger.addHandler(fh)

get_instance_logger(instance_name: str, host: str = '', port: int = 0, uid: str = '') -> LoggerAdapterT

Get an adapted logger instance for a given instance (driver/channel/transport)

Parameters:

Name Type Description Default
instance_name str

logger/instance name, i.e. "scrapli.driver"

 required
host str

host to add to logging extras if applicable

 ''
port int

port to add to logging extras if applicable

 0
uid str

unique id for a logging instance

 ''

Returns:

Name Type Description
LoggerAdapterT LoggerAdapterT

adapter logger for the instance
Source code in scrapli/logging.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def get_instance_logger(
    instance_name: str, host: str = "", port: int = 0, uid: str = ""
) -> LoggerAdapterT:
    """
    Get an adapted logger instance for a given instance (driver/channel/transport)

    Args:
        instance_name: logger/instance name, i.e. "scrapli.driver"
        host: host to add to logging extras if applicable
        port: port to add to logging extras if applicable
        uid: unique id for a logging instance

    Returns:
        LoggerAdapterT: adapter logger for the instance

    Raises:
        N/A

    """
    extras = {}

    if host and port:
        extras["host"] = host
        extras["port"] = str(port)

    if uid:
        extras["uid"] = uid

    _logger = getLogger(instance_name)
    return LoggerAdapter(_logger, extra=extras)