YANG Connector

26.3 · active · verified Thu Apr 16

YANG Connector is a Python package providing tools for testing YANG models via NETCONF and RESTCONF protocols. Primarily, it offers a NETCONF client by wrapping the `ncclient` library and integrates seamlessly with the pyATS framework. The library is actively maintained, with the current version being 26.3, and supports NETCONF v1.0 and v1.1.

Common errors

```
Exception: configure is not a supported method of this Netconf class, since a more suitable method, edit_config, is recommended.
```
cause Attempting to use the `configure()` method, which is not implemented for `yang.connector.Netconf`.

fix Replace calls to `device.netconf.configure(msg)` with `device.netconf.edit_config(target='running', config=msg)` (or `candidate`).
```
ncclient timed out while waiting for an rpc-reply.
```
cause The network device took too long to respond to an RPC request, exceeding the configured timeout.

fix When connecting or calling `request()`, set a larger `timeout` value: `device.connect(timeout=60)` or `device.netconf.request(rpc_message, timeout=120)`.
```
ncclient.operations.errors.OperationError: Missing <message-id> attribute in rpc-reply.
```
cause The NETCONF server's reply did not include a `message-id` attribute or it did not match the sent request's `message-id`.

fix This often indicates a non-compliant NETCONF server or an issue with how the RPC request was formulated if done manually. Verify server compliance or ensure `ncclient` is correctly handling RPC message IDs.

Warnings

breaking The `configure()` method in `yang.connector.Netconf` is a placeholder and will raise an exception. It is not supported for NETCONF operations.
Fix: Use the `edit_config()` method instead for sending configuration changes. The `request()` method can also be used for raw RPC requests.
gotcha When using `request()` or other methods, if the NETCONF `rpc-reply` is in a wrong format, or if `message-id` is missing/mismatched, `ncclient` (which `yang.connector` wraps) may raise an `OperationError`.
Fix: Ensure that the device's RPC replies conform to expected NETCONF standards and that message-ids are correctly handled if manually constructed. Review `ncclient` documentation for error handling.
gotcha The `request()` method in `Netconf` can raise a `TimeoutExpiredError` if the device does not send an `rpc-reply` within the specified timeout duration.
Fix: Increase the `timeout` parameter when establishing the connection or when making the `request()` call, especially for large queries or slow devices. The default timeout is 30 seconds.

Install

pip install yang.connector Install stable version

Imports

Netconf
```
from yang.connector import Netconf
```
Gnmi
```
from yang.connector.gnmi import Gnmi
```
For GNMI client functionality, import from the 'gnmi' submodule.

Quickstart

This quickstart demonstrates how to connect to a network device using `yang.connector.Netconf` via the pyATS testbed loader. It then retrieves a filtered configuration (e.g., interfaces) and lists the device's NETCONF capabilities. Credentials should be provided via environment variables for security.

import os
from pyats.topology import loader

# Define device connection details using environment variables for security
DEVICE_IP = os.environ.get('YANG_CONNECTOR_DEVICE_IP', 'your_device_ip')
DEVICE_PORT = int(os.environ.get('YANG_CONNECTOR_DEVICE_PORT', 830)) # Default NETCONF port
USERNAME = os.environ.get('YANG_CONNECTOR_USERNAME', 'admin')
PASSWORD = os.environ.get('YANG_CONNECTOR_PASSWORD', 'admin')

# A minimal pyATS Testbed YAML string for demonstration
testbed_yaml = f'''
devices:
  my_device:
    os: iosxe
    connections:
      netconf:
        class: yang.connector.Netconf
        ip: {DEVICE_IP}
        port: {DEVICE_PORT}
        protocol: netconf
        credentials:
          default:
            username: {USERNAME}
            password: {PASSWORD}
'''

try:
    # Load the testbed from the YAML string
    testbed = loader.load(testbed_yaml)
    device = testbed.devices['my_device']

    # Establish connection
    device.connect(via='netconf', init_exec_alias=False)
    print(f"Successfully connected to {device.name} via NETCONF.")

    # Example: Get configuration using get_config()
    # For a full configuration, an empty filter can be used
    # For specific data, provide an XML filter string
    filter_xml = '<filter type="subtree"><interfaces/></filter>'
    config_data = device.netconf.get_config(source='running', filter=filter_xml)
    print("\nRetrieved configuration (filtered for interfaces):\n")
    print(config_data.data_xml)

    # Example: Get capabilities
    print("\nServer Capabilities:")
    for cap in device.netconf.server_capabilities:
        print(f"- {cap}")

except Exception as e:
    print(f"An error occurred: {e}")
finally:
    # Disconnect from the device
    if 'device' in locals() and device.is_connected('netconf'):
        device.disconnect(via='netconf')
        print(f"Disconnected from {device.name}.")

view raw JSON →