Skip to content

base_driver

scrapli.driver.generic.base_driver

BaseGenericDriver

Source code in driver/generic/base_driver.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
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
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
class BaseGenericDriver:
    @staticmethod
    def _pre_send_command(
        host: str, command: str, failed_when_contains: Optional[Union[str, List[str]]] = None
    ) -> Response:
        """
        Handle pre "send_command" tasks for consistency between sync/async versions

        Args:
            host: string name of the host
            command: string to send to device in privilege exec mode
            failed_when_contains: string or list of strings indicating failure if found in response

        Returns:
            Response: Scrapli Response object

        Raises:
            ScrapliTypeError: if command is anything but a string

        """
        if not isinstance(command, str):
            raise ScrapliTypeError(
                f"`send_command` expects a single string, got {type(command)}, "
                "to send a list of commands use the `send_commands` method instead."
            )

        return Response(
            host=host,
            channel_input=command,
            failed_when_contains=failed_when_contains,
        )

    @staticmethod
    def _post_send_command(
        raw_response: bytes, processed_response: bytes, response: Response
    ) -> Response:
        """
        Handle post "send_command" tasks for consistency between sync/async versions

        Args:
            raw_response: raw response returned from the channel
            processed_response: processed response returned from the channel
            response: response object to update with channel results

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        response.record_response(result=processed_response)
        response.raw_result = raw_response
        return response

    @staticmethod
    def _pre_send_commands(commands: List[str]) -> MultiResponse:
        """
        Handle pre "send_command" tasks for consistency between sync/async versions

        Args:
            commands: list of strings to send to device in privilege exec mode

        Returns:
            MultiResponse: Scrapli MultiResponse object

        Raises:
            ScrapliTypeError: if command is anything but a string

        """
        if not isinstance(commands, list):
            raise ScrapliTypeError(
                f"`send_commands` expects a list of strings, got {type(commands)}, "
                "to send a single command use the `send_command` method instead."
            )

        return MultiResponse()

    @staticmethod
    def _pre_send_from_file(file: str, caller: str) -> List[str]:
        """
        Handle pre "send_*_from_file" tasks for consistency between sync/async versions

        Args:
            file: string path to file
            caller: name of the calling method for more helpful error message

        Returns:
            list: list of commands/configs read from file

        Raises:
            ScrapliTypeError: if anything but a string is provided for `file`

        """
        if not isinstance(file, str):
            raise ScrapliTypeError(f"`{caller}` expects a string path to a file, got {type(file)}")
        resolved_file = resolve_file(file)

        with open(resolved_file, "r", encoding="utf-8") as f:
            commands = f.read().splitlines()

        return commands

    @classmethod
    def _pre_send_interactive(
        cls,
        host: str,
        interact_events: Union[List[Tuple[str, str]], List[Tuple[str, str, bool]]],
        failed_when_contains: Optional[Union[str, List[str]]] = None,
    ) -> Response:
        """
        Handle pre "send_interactive" tasks for consistency between sync/async versions

        Args:
            host: string name of the host
            interact_events: list of tuples containing the "interactions" with the device
                each list element must have an input and an expected response, and may have an
                optional bool for the third and final element -- the optional bool specifies if the
                input that is sent to the device is "hidden" (ex: password), if the hidden param is
                not provided it is assumed the input is "normal" (not hidden)
            failed_when_contains: string or list of strings indicating failure if found in response

        Returns:
            Response: Scrapli Response object

        Raises:
            N/A

        """
        joined_input = ", ".join(event[0] for event in interact_events)
        return cls._pre_send_command(
            host=host, command=joined_input, failed_when_contains=failed_when_contains
        )

ReadCallback

Source code in driver/generic/base_driver.py 
 25
 26
 27
 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
136
137
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
232
233
class ReadCallback:
    def __init__(
        self,
        callback: Callable[
            [Union["GenericDriver", "AsyncGenericDriver"], str],
            Union[None, Coroutine[Any, Any, None]],
        ],
        contains: str = "",
        not_contains: str = "",
        contains_re: str = "",
        case_insensitive: bool = True,
        multiline: bool = True,
        reset_output: bool = True,
        only_once: bool = False,
        next_delay: float = -1.0,
        next_timeout: float = -1.0,
        complete: bool = False,
        name: str = "",
    ):
        """
        Object representing a single callback to be used with `GenericDriver` `read_callback` method

        Though the callable is typed with GenericDriver and AsyncGenericDriver, the callable can of
        course accept a NetworkDriver or AsyncNetworkDriver or any class extending those, you just
        may get some IDE/mypy complaints!

        Args:
            callback: callback function to execute, callback function must accept instance of the
                class as first argument, and the  "read_output" as second
            contains: string of text that, if in the read output, indicates to execute this callback
            not_contains: string of text that should *not* be contained in the output
            contains_re: string of a regex pattern that will be compiled and used to match the
                callback
            case_insensitive: ignored unless contains_re provided, sets re.I on compiled regex
            multiline: ignored unless contains_re provided, sets re.M on compiled regex
            reset_output: bool indicating to reset (clear) the output or to pass the output along
                to the next iteration. Sometimes you may want to clear the output to not
                accidentally continue matching on one callback over and over again. You could also
                use `only_once` to help with that
            only_once: bool indicating if this callback should only ever be executed one time
            next_delay: optional sleep between reads for next callback check
            next_timeout: optionally set the transport timeout (to timeout the read operation) for
                the subsequent callback checks -- the default value of -1.0 will tell scrapli to use
                the "normal" transport timeout for the operation
            complete: bool indicating if this is the "last" callback to execute
            name: friendly name to give the callback, will be function name if not provided

        Returns:
            N/A

        Raises:
            N/A

        """
        self.name = name
        if self.name == "":
            self.name = callback.__name__

        self.callback = callback

        self.contains = contains
        self._contains_bytes = b""

        self.not_contains = not_contains
        self._not_contains_bytes = b""

        self.contains_re = contains_re
        self._contains_re_bytes: Optional[Pattern[bytes]] = None

        self.case_insensitive = case_insensitive
        self.multiline = multiline
        self.reset_output = reset_output

        self.only_once = only_once
        self._triggered = False

        self.next_delay = next_delay
        self.next_timeout = next_timeout

        self.complete = complete

        self._read_output = b""

    @property
    def contains_bytes(self) -> bytes:
        """
        Property to encode provided not contains if requested

        Args:
            N/A

        Returns:
            bytes: encoded not contains string

        Raises:
            N/A

        """
        if self.contains and not self._contains_bytes:
            self._contains_bytes = self.contains.encode()

        return self._contains_bytes

    @property
    def not_contains_bytes(self) -> bytes:
        """
        Property to encode provided contains if requested

        Args:
            N/A

        Returns:
            bytes: encoded contains string

        Raises:
            N/A

        """
        if self.not_contains and not self._not_contains_bytes:
            self._not_contains_bytes = self.not_contains.encode()

        return self._not_contains_bytes

    @property
    def contains_re_bytes(self) -> Pattern[bytes]:
        """
        Property to encode provided regex contains if requested

        Args:
            N/A

        Returns:
            re.Pattern: compiled bytes pattern

        Raises:
            N/A

        """
        if not self._contains_re_bytes:
            flags = 0

            if self.case_insensitive and self.multiline:
                flags = re.I | re.M
            elif self.case_insensitive:
                flags = re.I
            elif self.multiline:
                flags = re.M

            self._contains_re_bytes = re.compile(pattern=self.contains_re.encode(), flags=flags)

        return self._contains_re_bytes

    def check(self, read_output: bytes) -> bool:
        """
        Determine if a callback has matched based on device output

        Args:
            read_output: bytes read from the device

        Returns:
            bool: True/False indicating if the callback "matches" the output

        Raises:
            N/A

        """
        self._read_output = read_output

        if self.case_insensitive:
            _read_output = read_output.lower()
        else:
            _read_output = read_output

        if (
            self.contains_bytes
            and self.contains_bytes in _read_output
            and not (self.not_contains and self.not_contains_bytes in _read_output)
        ):
            return True

        if (
            self.contains_re
            and re.search(self.contains_re_bytes, _read_output)
            and not (self.not_contains and self.not_contains_bytes in _read_output)
        ):
            return True

        return False

    def run(
        self, driver: Union["GenericDriver", "AsyncGenericDriver"]
    ) -> Union[None, Awaitable[Any]]:
        """
        Run the callback

        Args:
            driver: driver object to pass to the callback function

        Returns:
            Union[None, Awaitable[Any]]: return the result of the callable if sync or the future

        Raises:
            N/A

        """
        if self.only_once is True:
            self._triggered = True

        return self.callback(driver, self._read_output.decode())

__init__(callback: Callable[[Union[GenericDriver, AsyncGenericDriver], str], Union[None, Coroutine[Any, Any, None]]], contains: str = '', not_contains: str = '', contains_re: str = '', case_insensitive: bool = True, multiline: bool = True, reset_output: bool = True, only_once: bool = False, next_delay: float = -1.0, next_timeout: float = -1.0, complete: bool = False, name: str = '')

Object representing a single callback to be used with GenericDriver read_callback method

Though the callable is typed with GenericDriver and AsyncGenericDriver, the callable can of course accept a NetworkDriver or AsyncNetworkDriver or any class extending those, you just may get some IDE/mypy complaints!

Parameters:

Name Type Description Default
callback Callable[[Union[GenericDriver, AsyncGenericDriver], str], Union[None, Coroutine[Any, Any, None]]]

callback function to execute, callback function must accept instance of the class as first argument, and the "read_output" as second

 required
contains str

string of text that, if in the read output, indicates to execute this callback

 ''
not_contains str

string of text that should not be contained in the output

 ''
contains_re str

string of a regex pattern that will be compiled and used to match the callback

 ''
case_insensitive bool

ignored unless contains_re provided, sets re.I on compiled regex

 True
multiline bool

ignored unless contains_re provided, sets re.M on compiled regex

 True
reset_output bool

bool indicating to reset (clear) the output or to pass the output along to the next iteration. Sometimes you may want to clear the output to not accidentally continue matching on one callback over and over again. You could also use only_once to help with that

 True
only_once bool

bool indicating if this callback should only ever be executed one time

 False
next_delay float

optional sleep between reads for next callback check

 -1.0
next_timeout float

optionally set the transport timeout (to timeout the read operation) for the subsequent callback checks -- the default value of -1.0 will tell scrapli to use the "normal" transport timeout for the operation

 -1.0
complete bool

bool indicating if this is the "last" callback to execute

 False
name str

friendly name to give the callback, will be function name if not provided

 ''

Returns:

Type Description

N/A
Source code in driver/generic/base_driver.py 
 26
 27
 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
def __init__(
    self,
    callback: Callable[
        [Union["GenericDriver", "AsyncGenericDriver"], str],
        Union[None, Coroutine[Any, Any, None]],
    ],
    contains: str = "",
    not_contains: str = "",
    contains_re: str = "",
    case_insensitive: bool = True,
    multiline: bool = True,
    reset_output: bool = True,
    only_once: bool = False,
    next_delay: float = -1.0,
    next_timeout: float = -1.0,
    complete: bool = False,
    name: str = "",
):
    """
    Object representing a single callback to be used with `GenericDriver` `read_callback` method

    Though the callable is typed with GenericDriver and AsyncGenericDriver, the callable can of
    course accept a NetworkDriver or AsyncNetworkDriver or any class extending those, you just
    may get some IDE/mypy complaints!

    Args:
        callback: callback function to execute, callback function must accept instance of the
            class as first argument, and the  "read_output" as second
        contains: string of text that, if in the read output, indicates to execute this callback
        not_contains: string of text that should *not* be contained in the output
        contains_re: string of a regex pattern that will be compiled and used to match the
            callback
        case_insensitive: ignored unless contains_re provided, sets re.I on compiled regex
        multiline: ignored unless contains_re provided, sets re.M on compiled regex
        reset_output: bool indicating to reset (clear) the output or to pass the output along
            to the next iteration. Sometimes you may want to clear the output to not
            accidentally continue matching on one callback over and over again. You could also
            use `only_once` to help with that
        only_once: bool indicating if this callback should only ever be executed one time
        next_delay: optional sleep between reads for next callback check
        next_timeout: optionally set the transport timeout (to timeout the read operation) for
            the subsequent callback checks -- the default value of -1.0 will tell scrapli to use
            the "normal" transport timeout for the operation
        complete: bool indicating if this is the "last" callback to execute
        name: friendly name to give the callback, will be function name if not provided

    Returns:
        N/A

    Raises:
        N/A

    """
    self.name = name
    if self.name == "":
        self.name = callback.__name__

    self.callback = callback

    self.contains = contains
    self._contains_bytes = b""

    self.not_contains = not_contains
    self._not_contains_bytes = b""

    self.contains_re = contains_re
    self._contains_re_bytes: Optional[Pattern[bytes]] = None

    self.case_insensitive = case_insensitive
    self.multiline = multiline
    self.reset_output = reset_output

    self.only_once = only_once
    self._triggered = False

    self.next_delay = next_delay
    self.next_timeout = next_timeout

    self.complete = complete

    self._read_output = b""

check(read_output: bytes) -> bool

Determine if a callback has matched based on device output

Parameters:

Name Type Description Default
read_output bytes

bytes read from the device

 required

Returns:

Name Type Description
bool bool

True/False indicating if the callback "matches" the output
Source code in driver/generic/base_driver.py 
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
def check(self, read_output: bytes) -> bool:
    """
    Determine if a callback has matched based on device output

    Args:
        read_output: bytes read from the device

    Returns:
        bool: True/False indicating if the callback "matches" the output

    Raises:
        N/A

    """
    self._read_output = read_output

    if self.case_insensitive:
        _read_output = read_output.lower()
    else:
        _read_output = read_output

    if (
        self.contains_bytes
        and self.contains_bytes in _read_output
        and not (self.not_contains and self.not_contains_bytes in _read_output)
    ):
        return True

    if (
        self.contains_re
        and re.search(self.contains_re_bytes, _read_output)
        and not (self.not_contains and self.not_contains_bytes in _read_output)
    ):
        return True

    return False

contains_bytes() -> bytes property

Property to encode provided not contains if requested

Returns:

Name Type Description
bytes bytes

encoded not contains string
Source code in driver/generic/base_driver.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@property
def contains_bytes(self) -> bytes:
    """
    Property to encode provided not contains if requested

    Args:
        N/A

    Returns:
        bytes: encoded not contains string

    Raises:
        N/A

    """
    if self.contains and not self._contains_bytes:
        self._contains_bytes = self.contains.encode()

    return self._contains_bytes

contains_re_bytes() -> Pattern[bytes] property

Property to encode provided regex contains if requested

Returns:

Type Description
Pattern[bytes]

re.Pattern: compiled bytes pattern
Source code in driver/generic/base_driver.py 
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
@property
def contains_re_bytes(self) -> Pattern[bytes]:
    """
    Property to encode provided regex contains if requested

    Args:
        N/A

    Returns:
        re.Pattern: compiled bytes pattern

    Raises:
        N/A

    """
    if not self._contains_re_bytes:
        flags = 0

        if self.case_insensitive and self.multiline:
            flags = re.I | re.M
        elif self.case_insensitive:
            flags = re.I
        elif self.multiline:
            flags = re.M

        self._contains_re_bytes = re.compile(pattern=self.contains_re.encode(), flags=flags)

    return self._contains_re_bytes

not_contains_bytes() -> bytes property

Property to encode provided contains if requested

Returns:

Name Type Description
bytes bytes

encoded contains string
Source code in driver/generic/base_driver.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@property
def not_contains_bytes(self) -> bytes:
    """
    Property to encode provided contains if requested

    Args:
        N/A

    Returns:
        bytes: encoded contains string

    Raises:
        N/A

    """
    if self.not_contains and not self._not_contains_bytes:
        self._not_contains_bytes = self.not_contains.encode()

    return self._not_contains_bytes

run(driver: Union[GenericDriver, AsyncGenericDriver]) -> Union[None, Awaitable[Any]]

Run the callback

Parameters:

Name Type Description Default
driver Union[GenericDriver, AsyncGenericDriver]

driver object to pass to the callback function

 required

Returns:

Type Description
Union[None, Awaitable[Any]]

Union[None, Awaitable[Any]]: return the result of the callable if sync or the future
Source code in driver/generic/base_driver.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def run(
    self, driver: Union["GenericDriver", "AsyncGenericDriver"]
) -> Union[None, Awaitable[Any]]:
    """
    Run the callback

    Args:
        driver: driver object to pass to the callback function

    Returns:
        Union[None, Awaitable[Any]]: return the result of the callable if sync or the future

    Raises:
        N/A

    """
    if self.only_once is True:
        self._triggered = True

    return self.callback(driver, self._read_output.decode())