Skip to content

sync_driver

scrapli_netconf.driver.sync_driver

NetconfDriver

Bases: Driver, NetconfBaseDriver

Source code in driver/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
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
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
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
class NetconfDriver(Driver, NetconfBaseDriver):
    # kinda hate this but need to tell mypy that channel in netconf land is in fact a channel of
    # type `NetconfChannel`
    channel: NetconfChannel

    def __init__(
        self,
        host: str,
        port: int = 830,
        strip_namespaces: bool = False,
        strict_datastores: bool = False,
        auth_username: str = "",
        auth_password: str = "",
        auth_private_key: str = "",
        auth_private_key_passphrase: str = "",
        auth_strict_key: bool = True,
        auth_bypass: bool = False,
        timeout_socket: float = 15.0,
        timeout_transport: float = 30.0,
        timeout_ops: float = 30.0,
        comms_prompt_pattern: str = r"^[a-z0-9.\-@()/:]{1,48}[#>$]\s*$",
        comms_return_char: str = "\n",
        ssh_config_file: Union[str, bool] = False,
        ssh_known_hosts_file: Union[str, bool] = False,
        on_init: Optional[Callable[..., Any]] = None,
        on_open: Optional[Callable[..., Any]] = None,
        on_close: Optional[Callable[..., Any]] = None,
        transport: str = "system",
        transport_options: Optional[Dict[str, Any]] = None,
        channel_log: Union[str, bool] = False,
        channel_lock: bool = False,
        preferred_netconf_version: Optional[str] = None,
        use_compressed_parser: bool = True,
    ) -> None:
        super().__init__(
            host=host,
            port=port,
            auth_username=auth_username,
            auth_password=auth_password,
            auth_private_key=auth_private_key,
            auth_private_key_passphrase=auth_private_key_passphrase,
            auth_strict_key=auth_strict_key,
            auth_bypass=auth_bypass,
            timeout_socket=timeout_socket,
            timeout_transport=timeout_transport,
            timeout_ops=timeout_ops,
            comms_prompt_pattern=comms_prompt_pattern,
            comms_return_char=comms_return_char,
            ssh_config_file=ssh_config_file,
            ssh_known_hosts_file=ssh_known_hosts_file,
            on_init=on_init,
            on_open=on_open,
            on_close=on_close,
            transport=transport,
            transport_options=transport_options,
            channel_log=channel_log,
            channel_lock=channel_lock,
        )

        _preferred_netconf_version = self._determine_preferred_netconf_version(
            preferred_netconf_version=preferred_netconf_version
        )
        _preferred_xml_parser = self._determine_preferred_xml_parser(
            use_compressed_parser=use_compressed_parser
        )
        self._netconf_base_channel_args = NetconfBaseChannelArgs(
            netconf_version=_preferred_netconf_version, xml_parser=_preferred_xml_parser
        )

        self.channel = NetconfChannel(
            transport=self.transport,
            base_channel_args=self._base_channel_args,
            netconf_base_channel_args=self._netconf_base_channel_args,
        )

        self.strip_namespaces = strip_namespaces
        self.strict_datastores = strict_datastores
        self.server_capabilities: List[str] = []
        self.readable_datastores: List[str] = []
        self.writeable_datastores: List[str] = []
        self.message_id = 101

    def open(self) -> None:
        """
        Open netconf connection to server

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """
        self._pre_open_closing_log(closing=False)

        self.transport.open_netconf()

        # in the future this and scrapli core should just have a class attribute of the transports
        # that require this "in channel" auth so we can dynamically figure that out rather than
        # just look at the name of the transport
        if "system" in self.transport_name:
            self.channel.channel_authenticate_netconf(
                auth_password=self.auth_password,
                auth_private_key_passphrase=self.auth_private_key_passphrase,
            )

        self.channel.open_netconf()

        self._build_readable_datastores()
        self._build_writeable_datastores()

        self._post_open_closing_log(closing=False)

    def get(self, filter_: str, filter_type: str = "subtree") -> NetconfResponse:
        """
        Netconf get operation

        Args:
            filter_: filter to apply to the get
            filter_type: type of filter; subtree|xpath

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_get(filter_=filter_, filter_type=filter_type)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def get_config(
        self,
        source: str = "running",
        filter_: Optional[str] = None,
        filter_type: str = "subtree",
        default_type: Optional[str] = None,
    ) -> NetconfResponse:
        """
        Netconf get-config operation

        Args:
            source: configuration source to get; typically one of running|startup|candidate
            filter_: string of filter(s) to apply to configuration
            filter_type: type of filter; subtree|xpath
            default_type: string of with-default mode to apply when retrieving configuration

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_get_config(
            source=source, filter_=filter_, filter_type=filter_type, default_type=default_type
        )
        raw_response = self.channel.send_input_netconf(response.channel_input)

        response.record_response(raw_response)
        return response

    def edit_config(self, config: str, target: str = "running") -> NetconfResponse:
        """
        Netconf get-config operation

        Args:
            config: configuration to send to device
            target: configuration source to target; running|startup|candidate

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_edit_config(config=config, target=target)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def delete_config(self, target: str = "candidate") -> NetconfResponse:
        """
        Netconf delete-config operation

        Args:
            target: configuration source to target; startup|candidate

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_delete_config(target=target)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def commit(
        self,
        confirmed: bool = False,
        timeout: Optional[int] = None,
        persist: Optional[Union[int, str]] = None,
        persist_id: Optional[Union[int, str]] = None,
    ) -> NetconfResponse:
        """
        Netconf commit config operation

        Args:
            confirmed: whether this is a confirmed commit
            timeout: specifies the confirm timeout in seconds
            persist: make the confirmed commit survive a session termination, and set a token on
                the ongoing confirmed commit
            persist_id: value must be equal to the value given in the <persist> parameter to the
                original <commit> operation.

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_commit(
            confirmed=confirmed,
            timeout=timeout,
            persist=persist,
            persist_id=persist_id,
        )
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def discard(self) -> NetconfResponse:
        """
        Netconf discard config operation

        Args:
            N/A

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_discard()
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def lock(self, target: str) -> NetconfResponse:
        """
        Netconf lock operation

        Args:
            target: configuration source to target; running|startup|candidate

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_lock(target=target)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def unlock(self, target: str) -> NetconfResponse:
        """
        Netconf unlock operation

        Args:
            target: configuration source to target; running|startup|candidate

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_unlock(target=target)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def rpc(self, filter_: Union[str, _Element]) -> NetconfResponse:
        """
        Netconf "rpc" operation

        Typically used with juniper devices or if you want to build/send your own payload in a more
        manual fashion. You can provide a string that will be loaded as an lxml element, or you can
        provide an lxml element yourself.

        Args:
            filter_: filter/rpc to execute

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_rpc(filter_=filter_)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def validate(self, source: str) -> NetconfResponse:
        """
        Netconf "validate" operation

        Args:
            source: configuration source to validate; typically one of running|startup|candidate

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_validate(source=source)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    def copy_config(self, source: str, target: str) -> NetconfResponse:
        """
        Netconf "copy-config" operation

        Args:
            source: configuration, url, or datastore to copy into the target datastore
            target: destination to copy the source to

        Returns:
            NetconfResponse: scrapli_netconf NetconfResponse object

        Raises:
            N/A

        """
        response = self._pre_copy_config(source=source, target=target)
        raw_response = self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

commit(confirmed: bool = False, timeout: Optional[int] = None, persist: Optional[Union[int, str]] = None, persist_id: Optional[Union[int, str]] = None) -> NetconfResponse

Netconf commit config operation

Parameters:

Name Type Description Default
confirmed bool

whether this is a confirmed commit

 False
timeout Optional[int]

specifies the confirm timeout in seconds

 None
persist Optional[Union[int, str]]

make the confirmed commit survive a session termination, and set a token on the ongoing confirmed commit

 None
persist_id Optional[Union[int, str]]

value must be equal to the value given in the parameter to the original operation.

 None

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
def commit(
    self,
    confirmed: bool = False,
    timeout: Optional[int] = None,
    persist: Optional[Union[int, str]] = None,
    persist_id: Optional[Union[int, str]] = None,
) -> NetconfResponse:
    """
    Netconf commit config operation

    Args:
        confirmed: whether this is a confirmed commit
        timeout: specifies the confirm timeout in seconds
        persist: make the confirmed commit survive a session termination, and set a token on
            the ongoing confirmed commit
        persist_id: value must be equal to the value given in the <persist> parameter to the
            original <commit> operation.

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_commit(
        confirmed=confirmed,
        timeout=timeout,
        persist=persist,
        persist_id=persist_id,
    )
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

copy_config(source: str, target: str) -> NetconfResponse

Netconf "copy-config" operation

Parameters:

Name Type Description Default
source str

configuration, url, or datastore to copy into the target datastore

 required
target str

destination to copy the source to

 required

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
def copy_config(self, source: str, target: str) -> NetconfResponse:
    """
    Netconf "copy-config" operation

    Args:
        source: configuration, url, or datastore to copy into the target datastore
        target: destination to copy the source to

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_copy_config(source=source, target=target)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

delete_config(target: str = 'candidate') -> NetconfResponse

Netconf delete-config operation

Parameters:

Name Type Description Default
target str

configuration source to target; startup|candidate

 'candidate'

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def delete_config(self, target: str = "candidate") -> NetconfResponse:
    """
    Netconf delete-config operation

    Args:
        target: configuration source to target; startup|candidate

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_delete_config(target=target)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

discard() -> NetconfResponse

Netconf discard config operation

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def discard(self) -> NetconfResponse:
    """
    Netconf discard config operation

    Args:
        N/A

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_discard()
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

edit_config(config: str, target: str = 'running') -> NetconfResponse

Netconf get-config operation

Parameters:

Name Type Description Default
config str

configuration to send to device

 required
target str

configuration source to target; running|startup|candidate

 'running'

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
def edit_config(self, config: str, target: str = "running") -> NetconfResponse:
    """
    Netconf get-config operation

    Args:
        config: configuration to send to device
        target: configuration source to target; running|startup|candidate

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_edit_config(config=config, target=target)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

get(filter_: str, filter_type: str = 'subtree') -> NetconfResponse

Netconf get operation

Parameters:

Name Type Description Default
filter_ str

filter to apply to the get

 required
filter_type str

type of filter; subtree|xpath

 'subtree'

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get(self, filter_: str, filter_type: str = "subtree") -> NetconfResponse:
    """
    Netconf get operation

    Args:
        filter_: filter to apply to the get
        filter_type: type of filter; subtree|xpath

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_get(filter_=filter_, filter_type=filter_type)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

get_config(source: str = 'running', filter_: Optional[str] = None, filter_type: str = 'subtree', default_type: Optional[str] = None) -> NetconfResponse

Netconf get-config operation

Parameters:

Name Type Description Default
source str

configuration source to get; typically one of running|startup|candidate

 'running'
filter_ Optional[str]

string of filter(s) to apply to configuration

 None
filter_type str

type of filter; subtree|xpath

 'subtree'
default_type Optional[str]

string of with-default mode to apply when retrieving configuration

 None

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
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
def get_config(
    self,
    source: str = "running",
    filter_: Optional[str] = None,
    filter_type: str = "subtree",
    default_type: Optional[str] = None,
) -> NetconfResponse:
    """
    Netconf get-config operation

    Args:
        source: configuration source to get; typically one of running|startup|candidate
        filter_: string of filter(s) to apply to configuration
        filter_type: type of filter; subtree|xpath
        default_type: string of with-default mode to apply when retrieving configuration

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_get_config(
        source=source, filter_=filter_, filter_type=filter_type, default_type=default_type
    )
    raw_response = self.channel.send_input_netconf(response.channel_input)

    response.record_response(raw_response)
    return response

lock(target: str) -> NetconfResponse

Netconf lock operation

Parameters:

Name Type Description Default
target str

configuration source to target; running|startup|candidate

 required

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def lock(self, target: str) -> NetconfResponse:
    """
    Netconf lock operation

    Args:
        target: configuration source to target; running|startup|candidate

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_lock(target=target)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

open() -> None

Open netconf connection to server

Returns:

Type Description
None

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

    Args:
        N/A

    Returns:
        None

    Raises:
        N/A

    """
    self._pre_open_closing_log(closing=False)

    self.transport.open_netconf()

    # in the future this and scrapli core should just have a class attribute of the transports
    # that require this "in channel" auth so we can dynamically figure that out rather than
    # just look at the name of the transport
    if "system" in self.transport_name:
        self.channel.channel_authenticate_netconf(
            auth_password=self.auth_password,
            auth_private_key_passphrase=self.auth_private_key_passphrase,
        )

    self.channel.open_netconf()

    self._build_readable_datastores()
    self._build_writeable_datastores()

    self._post_open_closing_log(closing=False)

rpc(filter_: Union[str, _Element]) -> NetconfResponse

Netconf "rpc" operation

Typically used with juniper devices or if you want to build/send your own payload in a more manual fashion. You can provide a string that will be loaded as an lxml element, or you can provide an lxml element yourself.

Parameters:

Name Type Description Default
filter_ Union[str, _Element]

filter/rpc to execute

 required

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
def rpc(self, filter_: Union[str, _Element]) -> NetconfResponse:
    """
    Netconf "rpc" operation

    Typically used with juniper devices or if you want to build/send your own payload in a more
    manual fashion. You can provide a string that will be loaded as an lxml element, or you can
    provide an lxml element yourself.

    Args:
        filter_: filter/rpc to execute

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_rpc(filter_=filter_)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

unlock(target: str) -> NetconfResponse

Netconf unlock operation

Parameters:

Name Type Description Default
target str

configuration source to target; running|startup|candidate

 required

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def unlock(self, target: str) -> NetconfResponse:
    """
    Netconf unlock operation

    Args:
        target: configuration source to target; running|startup|candidate

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_unlock(target=target)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

validate(source: str) -> NetconfResponse

Netconf "validate" operation

Parameters:

Name Type Description Default
source str

configuration source to validate; typically one of running|startup|candidate

 required

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/sync_driver.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def validate(self, source: str) -> NetconfResponse:
    """
    Netconf "validate" operation

    Args:
        source: configuration source to validate; typically one of running|startup|candidate

    Returns:
        NetconfResponse: scrapli_netconf NetconfResponse object

    Raises:
        N/A

    """
    response = self._pre_validate(source=source)
    raw_response = self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response