Skip to content

async_driver

scrapli_netconf.driver.async_driver

AsyncNetconfDriver

Bases: AsyncDriver, NetconfBaseDriver

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

    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 = AsyncNetconfChannel(
            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

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

        Args:
            N/A

        Returns:
            None

        Raises:
            N/A

        """
        self._pre_open_closing_log(closing=False)

        await self.transport.open_netconf()
        await self.channel.open_netconf()

        self._build_readable_datastores()
        self._build_writeable_datastores()

        self._post_open_closing_log(closing=False)

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

        Args:
            filter_: string 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)

        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await self.channel.send_input_netconf(response.channel_input)
        response.record_response(raw_response)
        return response

    async 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 = await 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 async

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/async_driver.py 
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
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

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

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/async_driver.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

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

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/async_driver.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

discard() -> NetconfResponse async

Netconf discard config operation

Returns:

Name Type Description
NetconfResponse NetconfResponse

scrapli_netconf NetconfResponse object
Source code in driver/async_driver.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

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

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/async_driver.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

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

Netconf get operation

Parameters:

Name Type Description Default
filter_ str

string 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/async_driver.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def get(self, filter_: str, filter_type: str = "subtree") -> NetconfResponse:
    """
    Netconf get operation

    Args:
        filter_: string 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 = await 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 async

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/async_driver.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
async 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 = await self.channel.send_input_netconf(response.channel_input)

    response.record_response(raw_response)
    return response

lock(target: str) -> NetconfResponse async

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/async_driver.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

open() -> None async

Open netconf connection to server

Returns:

Type Description
None

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

    Args:
        N/A

    Returns:
        None

    Raises:
        N/A

    """
    self._pre_open_closing_log(closing=False)

    await self.transport.open_netconf()
    await 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 async

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/async_driver.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

unlock(target: str) -> NetconfResponse async

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/async_driver.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response

validate(source: str) -> NetconfResponse async

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/async_driver.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
async 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 = await self.channel.send_input_netconf(response.channel_input)
    response.record_response(raw_response)
    return response