Skip to content

sync_driver

scrapli.driver.base.sync_driver

Driver

Bases: BaseDriver

Source code in driver/base/sync_driver.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 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
class Driver(BaseDriver):
    def __init__(self, **kwargs: Any):
        super().__init__(**kwargs)

        if self.transport_name in ASYNCIO_TRANSPORTS:
            raise ScrapliValueError(
                "provided transport is *not* an sync transport, must use an sync transport with"
                " the (sync)Driver(s)"
            )

        self.channel = Channel(
            transport=self.transport,
            base_channel_args=self._base_channel_args,
        )

        # ensure the "on_init" func is called in the actual driver (sync/async rather than the base
        # class) since we want to wait until after the channel is setup!
        if self.on_init:
            self.on_init(self)

    def __enter__(self: _T) -> _T:
        """
        Enter method for context manager

        Args:
            N/A

        Returns:
            _T: a concrete implementation of the opened Driver object

        Raises:
            N/A

        """
        self.open()
        return self

    def __exit__(
        self,
        exception_type: Optional[Type[BaseException]],
        exception_value: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        """
        Exit method to cleanup for context manager

        Args:
            exception_type: exception type being raised
            exception_value: message from exception being raised
            traceback: traceback from exception being raised

        Returns:
            None

        Raises:
            N/A

        """
        self.close()

    def open(self) -> None:
        """
        Open the scrapli connection

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """
        self._pre_open_closing_log(closing=False)

        self.transport.open()
        self.channel.open()

        if self.transport_name in ("system",) and not self.auth_bypass:
            self.channel.channel_authenticate_ssh(
                auth_password=self.auth_password,
                auth_private_key_passphrase=self.auth_private_key_passphrase,
            )
        if (
            self.transport_name
            in (
                "telnet",
                "asynctelnet",
            )
            and not self.auth_bypass
        ):
            self.channel.channel_authenticate_telnet(
                auth_username=self.auth_username, auth_password=self.auth_password
            )

        if self.on_open:
            self.on_open(self)

        self._post_open_closing_log(closing=False)

    def close(self) -> None:
        """
        Close the scrapli connection

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """
        self._pre_open_closing_log(closing=True)

        if self.on_close:
            self.on_close(self)

        self.transport.close()
        self.channel.close()

        self._post_open_closing_log(closing=True)

    def commandeer(self, conn: "Driver", execute_on_open: bool = True) -> None:
        """
        Commandeer an existing connection

        Used to "take over" or "commandeer" a connection. This method accepts a second scrapli conn
        object and "steals" the transport from this connection and uses it for the current instance.
        The primary reason you would want this is to use a `GenericDriver` to connect to a console
        server and then to "commandeer" that connection and convert it to a "normal" network driver
        connection type (i.e. Junos, EOS, etc.) once connected to the network device (via the
        console server).

        Right now closing the connection that "commandeers" the initial connection will *also close
        the original connection* -- this is because we are re-using the transport in this new conn.
        In the future perhaps this will change to *not* close the original connection so users can
        handle any type of cleanup operations that need to happen on the original connection.
        Alternatively, you can simply continue using the "original" connection to close things for
        yourself or do any type of clean up work (just dont close the commandeering connection!).

        Args:
            conn: connection to commandeer
            execute_on_open: execute the `on_open` function of the current object once the existing
                connection has been commandeered

        Returns:
            None

        Raises:
            N/A

        """
        original_logger = conn.logger
        original_transport = conn.transport
        original_transport_logger = conn.transport.logger
        original_channel_logger = conn.channel.logger
        original_channel_channel_log = conn.channel.channel_log

        self.logger = original_logger
        self.channel.logger = original_channel_logger
        self.channel.transport = original_transport
        self.transport = original_transport
        self.transport.logger = original_transport_logger

        if original_channel_channel_log is not None:
            # if the original connection had a channel log we also commandeer that; note that when
            # the new connection is closed this will also close the channel log; see docstring.
            self.channel.channel_log = original_channel_channel_log

        if execute_on_open and self.on_open is not None:
            self.on_open(self)

__enter__() -> _T

Enter method for context manager

Returns:

Name Type Description
_T _T

a concrete implementation of the opened Driver object
Source code in driver/base/sync_driver.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __enter__(self: _T) -> _T:
    """
    Enter method for context manager

    Args:
        N/A

    Returns:
        _T: a concrete implementation of the opened Driver object

    Raises:
        N/A

    """
    self.open()
    return self

__exit__(exception_type: Optional[Type[BaseException]], exception_value: Optional[BaseException], traceback: Optional[TracebackType]) -> None

Exit method to cleanup for context manager

Parameters:

Name Type Description Default
exception_type Optional[Type[BaseException]]

exception type being raised

 required
exception_value Optional[BaseException]

message from exception being raised

 required
traceback Optional[TracebackType]

traceback from exception being raised

 required

Returns:

Type Description
None

None
Source code in driver/base/sync_driver.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def __exit__(
    self,
    exception_type: Optional[Type[BaseException]],
    exception_value: Optional[BaseException],
    traceback: Optional[TracebackType],
) -> None:
    """
    Exit method to cleanup for context manager

    Args:
        exception_type: exception type being raised
        exception_value: message from exception being raised
        traceback: traceback from exception being raised

    Returns:
        None

    Raises:
        N/A

    """
    self.close()

close() -> None

Close the scrapli connection

Returns:

Type Description
None

None
Source code in driver/base/sync_driver.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def close(self) -> None:
    """
    Close the scrapli connection

    Args:
        N/A

    Returns:
        None

    Raises:
        N/A

    """
    self._pre_open_closing_log(closing=True)

    if self.on_close:
        self.on_close(self)

    self.transport.close()
    self.channel.close()

    self._post_open_closing_log(closing=True)

commandeer(conn: Driver, execute_on_open: bool = True) -> None

Commandeer an existing connection

Used to "take over" or "commandeer" a connection. This method accepts a second scrapli conn object and "steals" the transport from this connection and uses it for the current instance. The primary reason you would want this is to use a GenericDriver to connect to a console server and then to "commandeer" that connection and convert it to a "normal" network driver connection type (i.e. Junos, EOS, etc.) once connected to the network device (via the console server).

Right now closing the connection that "commandeers" the initial connection will also close the original connection -- this is because we are re-using the transport in this new conn. In the future perhaps this will change to not close the original connection so users can handle any type of cleanup operations that need to happen on the original connection. Alternatively, you can simply continue using the "original" connection to close things for yourself or do any type of clean up work (just dont close the commandeering connection!).

Parameters:

Name Type Description Default
conn Driver

connection to commandeer

 required
execute_on_open bool

execute the on_open function of the current object once the existing connection has been commandeered

 True

Returns:

Type Description
None

None
Source code in driver/base/sync_driver.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
def commandeer(self, conn: "Driver", execute_on_open: bool = True) -> None:
    """
    Commandeer an existing connection

    Used to "take over" or "commandeer" a connection. This method accepts a second scrapli conn
    object and "steals" the transport from this connection and uses it for the current instance.
    The primary reason you would want this is to use a `GenericDriver` to connect to a console
    server and then to "commandeer" that connection and convert it to a "normal" network driver
    connection type (i.e. Junos, EOS, etc.) once connected to the network device (via the
    console server).

    Right now closing the connection that "commandeers" the initial connection will *also close
    the original connection* -- this is because we are re-using the transport in this new conn.
    In the future perhaps this will change to *not* close the original connection so users can
    handle any type of cleanup operations that need to happen on the original connection.
    Alternatively, you can simply continue using the "original" connection to close things for
    yourself or do any type of clean up work (just dont close the commandeering connection!).

    Args:
        conn: connection to commandeer
        execute_on_open: execute the `on_open` function of the current object once the existing
            connection has been commandeered

    Returns:
        None

    Raises:
        N/A

    """
    original_logger = conn.logger
    original_transport = conn.transport
    original_transport_logger = conn.transport.logger
    original_channel_logger = conn.channel.logger
    original_channel_channel_log = conn.channel.channel_log

    self.logger = original_logger
    self.channel.logger = original_channel_logger
    self.channel.transport = original_transport
    self.transport = original_transport
    self.transport.logger = original_transport_logger

    if original_channel_channel_log is not None:
        # if the original connection had a channel log we also commandeer that; note that when
        # the new connection is closed this will also close the channel log; see docstring.
        self.channel.channel_log = original_channel_channel_log

    if execute_on_open and self.on_open is not None:
        self.on_open(self)

open() -> None

Open the scrapli connection

Returns:

Type Description
None

None
Source code in driver/base/sync_driver.py 
 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
def open(self) -> None:
    """
    Open the scrapli connection

    Args:
        N/A

    Returns:
        None

    Raises:
        N/A

    """
    self._pre_open_closing_log(closing=False)

    self.transport.open()
    self.channel.open()

    if self.transport_name in ("system",) and not self.auth_bypass:
        self.channel.channel_authenticate_ssh(
            auth_password=self.auth_password,
            auth_private_key_passphrase=self.auth_private_key_passphrase,
        )
    if (
        self.transport_name
        in (
            "telnet",
            "asynctelnet",
        )
        and not self.auth_bypass
    ):
        self.channel.channel_authenticate_telnet(
            auth_username=self.auth_username, auth_password=self.auth_password
        )

    if self.on_open:
        self.on_open(self)

    self._post_open_closing_log(closing=False)