tessl/pypi-ncclient
Python library for NETCONF clients with support for SSH, TLS, and device-specific operations
—
Pending
netconf-operations.mddocs/
NETCONF Operations
Complete implementation of NETCONF standard operations (RFC 6241) for configuration management, datastore operations, and session control. Operations are accessible through Manager instances and support both synchronous and asynchronous execution modes.
Capabilities
Configuration Retrieval
Operations for retrieving configuration and operational data from NETCONF datastores.
def get_config(source, filter=None, **kwargs):
"""
Retrieve configuration from specified datastore.
Parameters:
- source: str, datastore name ('running', 'candidate', 'startup')
- filter: tuple or str, filter specification (subtree or xpath)
Returns:
RPCReply: Configuration data response
"""
def get(filter=None, **kwargs):
"""
Retrieve configuration and state data.
Parameters:
- filter: tuple or str, filter specification (subtree or xpath)
Returns:
RPCReply: Combined configuration and operational data
"""
def get_schema(identifier, version=None, format=None):
"""
Retrieve YANG schema from server.
Parameters:
- identifier: str, schema identifier
- version: str, schema version (optional)
- format: str, schema format (yang, yin, rng, rnc)
Returns:
RPCReply: Schema content
"""Configuration Editing
Operations for modifying device configuration through NETCONF datastores.
def edit_config(target, config, default_operation=None,
test_option=None, error_option=None):
"""
Edit configuration in target datastore.
Parameters:
- target: str, target datastore ('running', 'candidate', 'startup')
- config: str or Element, configuration data
- default_operation: str, default operation ('merge', 'replace', 'none')
- test_option: str, test option ('test-then-set', 'set', 'test-only')
- error_option: str, error handling ('stop-on-error', 'continue-on-error', 'rollback-on-error')
Returns:
RPCReply: Edit operation result
"""
def copy_config(source, target):
"""
Copy configuration between datastores.
Parameters:
- source: str, source datastore or configuration
- target: str, target datastore
Returns:
RPCReply: Copy operation result
"""
def delete_config(target):
"""
Delete configuration datastore.
Parameters:
- target: str, target datastore (not 'running')
Returns:
RPCReply: Delete operation result
"""Transaction Management
Operations for managing configuration transactions and validation.
def validate(source):
"""
Validate configuration datastore.
Parameters:
- source: str, datastore to validate ('candidate', 'running', 'startup')
Returns:
RPCReply: Validation result
"""
def commit(confirmed=False, timeout=None, persist=None,
persist_id=None):
"""
Commit candidate configuration to running datastore.
Parameters:
- confirmed: bool, confirmed commit
- timeout: int, confirm timeout in seconds
- persist: str, persist ID for confirmed commit
- persist_id: str, existing persist ID to confirm
Returns:
RPCReply: Commit operation result
"""
def discard_changes():
"""
Discard changes in candidate datastore.
Returns:
RPCReply: Discard operation result
"""
def cancel_commit(persist_id=None):
"""
Cancel confirmed commit.
Parameters:
- persist_id: str, persist ID to cancel (optional)
Returns:
RPCReply: Cancel operation result
"""Datastore Locking
Operations for exclusive access control to configuration datastores.
def lock(target):
"""
Lock datastore for exclusive access.
Parameters:
- target: str, datastore to lock ('running', 'candidate', 'startup')
Returns:
RPCReply: Lock operation result
"""
def unlock(target):
"""
Unlock previously locked datastore.
Parameters:
- target: str, datastore to unlock
Returns:
RPCReply: Unlock operation result
"""Session Management
Operations for NETCONF session lifecycle management.
def close_session():
"""
Close current NETCONF session gracefully.
Returns:
RPCReply: Session close result
"""
def kill_session(session_id):
"""
Forcefully terminate another NETCONF session.
Parameters:
- session_id: str, session ID to terminate
Returns:
RPCReply: Kill session result
"""Notification Subscription
Operations for subscribing to NETCONF event notifications.
def create_subscription(filter=None, start_time=None,
stop_time=None):
"""
Create notification subscription.
Parameters:
- filter: str or tuple, notification filter
- start_time: str, subscription start time (ISO 8601)
- stop_time: str, subscription stop time (ISO 8601)
Returns:
RPCReply: Subscription creation result
"""Generic RPC Operations
Direct RPC execution for custom or vendor-specific operations.
def dispatch(rpc_command, source=None, filter=None):
"""
Execute arbitrary RPC command.
Parameters:
- rpc_command: str or Element, RPC command
- source: str, source datastore (for data operations)
- filter: str or tuple, data filter
Returns:
RPCReply: RPC execution result
"""
def rpc(op):
"""
Execute generic RPC operation.
Parameters:
- op: Element, RPC operation element
Returns:
RPCReply: RPC response
"""Device Control Operations
Additional operations for device management and control.
def poweroff_machine():
"""
Power off the device.
Returns:
RPCReply: Poweroff operation result
"""
def reboot_machine():
"""
Reboot the device.
Returns:
RPCReply: Reboot operation result
"""Core Operation Classes
class RPC:
"""Base class for NETCONF RPC operations."""
def request(self, *args, **kwargs):
"""Execute RPC request with parameters."""
class RPCReply:
"""Represents NETCONF RPC reply with result data."""
@property
def ok(self):
"""bool: Whether operation succeeded."""
@property
def data(self):
"""Element: Reply data content."""
@property
def data_xml(self):
"""str: Reply data as XML string."""
@property
def errors(self):
"""list: List of RPC errors."""
class RPCError(Exception):
"""Exception for NETCONF RPC errors."""
@property
def type(self):
"""str: Error type."""
@property
def tag(self):
"""str: Error tag."""
@property
def severity(self):
"""str: Error severity."""
@property
def message(self):
"""str: Error message."""
class RaiseMode:
"""Constants for RPC error raising behavior."""
NONE = 0 # Never raise exceptions
ERRORS = 1 # Raise for errors only
ALL = 2 # Raise for errors and warnings (default)Lock Context Manager
class LockContext:
"""Context manager for datastore locking."""
def __init__(self, session, device_handler, target):
"""
Initialize lock context.
Parameters:
- session: transport session
- device_handler: device handler
- target: str, datastore to lock
"""
def __enter__(self):
"""Acquire datastore lock."""
def __exit__(self, *args):
"""Release datastore lock."""Usage Examples
Basic Configuration Operations
from ncclient import manager
with manager.connect_ssh(host='device.example.com',
username='admin', password='admin') as m:
# Get running configuration
config = m.get_config(source='running')
print(config.data_xml)
# Edit configuration
new_config = '''
<config>
<interface xmlns="urn:example:interface">
<name>eth0</name>
<description>Updated interface</description>
</interface>
</config>
'''
result = m.edit_config(target='candidate', config=new_config)
if result.ok:
# Validate and commit
m.validate(source='candidate')
m.commit()
print("Configuration updated successfully")Using Datastore Locking
from ncclient import manager
with manager.connect_ssh(host='device.example.com',
username='admin', password='admin') as m:
# Use lock context manager
with m.locked('candidate'):
# Exclusive access to candidate datastore
config = '''
<config>
<system xmlns="urn:example:system">
<hostname>new-hostname</hostname>
</system>
</config>
'''
m.edit_config(target='candidate', config=config)
m.validate(source='candidate')
m.commit()
# Lock automatically releasedFiltering Data
from ncclient import manager
with manager.connect_ssh(host='device.example.com',
username='admin', password='admin') as m:
# Subtree filter
interface_filter = '''
<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces">
<interface>
<name/>
<type/>
<admin-status/>
</interface>
</interfaces>
'''
# Get filtered configuration
result = m.get_config(source='running',
filter=('subtree', interface_filter))
# XPath filter
xpath_filter = "/interfaces/interface[name='eth0']"
result = m.get_config(source='running',
filter=('xpath', xpath_filter))Error Handling
from ncclient import manager
from ncclient.operations import RPCError, RaiseMode
with manager.connect_ssh(host='device.example.com',
username='admin', password='admin') as m:
# Configure error handling
m.raise_mode = RaiseMode.ERRORS
try:
result = m.edit_config(target='running', config=invalid_config)
except RPCError as e:
print(f"RPC Error: {e.message}")
print(f"Error type: {e.type}")
print(f"Error tag: {e.tag}")Install with Tessl CLI
npx tessl i tessl/pypi-ncclient