tessl/pypi-ncclient
Python library for NETCONF clients with support for SSH, TLS, and device-specific operations
—
Pending
connection-management.mddocs/
Connection Management
Functions for establishing and managing NETCONF sessions over various transport protocols. NCClient provides flexible connection methods with support for device-specific parameters, authentication methods, and transport customization.
Capabilities
Generic Connection
Automatically selects appropriate transport (typically SSH) based on parameters provided.
def connect(host=None, port=830, username=None, password=None,
key_filename=None, timeout=None, device_params=None,
manager_params=None, nc_params=None, errors_params=None,
**kwargs):
"""
Initialize a Manager over automatically selected transport.
Parameters:
- host: str, hostname or IP address
- port: int, NETCONF port (default 830)
- username: str, username for authentication
- password: str, password for authentication
- key_filename: str, path to private key file
- timeout: float, connection timeout in seconds
- device_params: dict, device-specific parameters
- manager_params: dict, manager configuration
- nc_params: dict, NETCONF protocol parameters
- errors_params: dict, error handling configuration
- hostkey_verify: bool, verify host key (default True)
- allow_agent: bool, use SSH agent (default True)
- look_for_keys: bool, look for key files (default True)
- use_libssh: bool, use LibSSH transport (default False)
Returns:
Manager: NETCONF session manager instance
"""SSH Connection
Establishes NETCONF session over SSH transport with full SSH client capabilities.
def connect_ssh(host, port=830, username=None, password=None,
key_filename=None, timeout=None, allow_agent=True,
hostkey_verify=True, look_for_keys=True,
sock=None, **kwargs):
"""
Initialize a Manager over SSH transport.
Parameters:
- host: str, hostname or IP address
- port: int, SSH port (default 830)
- username: str, SSH username
- password: str, SSH password
- key_filename: str or list, SSH private key file(s)
- timeout: float, connection timeout
- allow_agent: bool, use SSH agent for authentication
- hostkey_verify: bool, verify SSH host key
- look_for_keys: bool, search for SSH keys in standard locations
- sock: socket, existing socket connection
Returns:
Manager: SSH-based NETCONF session manager
"""LibSSH Connection
Alternative SSH implementation using ssh-python library for enhanced performance.
def connect_libssh(host, port=830, username=None, password=None,
**kwargs):
"""
Initialize a Manager over LibSSH transport.
Requires 'ssh-python' package to be installed.
Parameters:
- host: str, hostname or IP address
- port: int, SSH port (default 830)
- username: str, SSH username
- password: str, SSH password
Returns:
Manager: LibSSH-based NETCONF session manager
"""TLS Connection
Establishes NETCONF session over TLS transport for secure communication.
def connect_tls(host, port=6513, certfile=None, keyfile=None,
ca_certs=None, cert_reqs=None, ssl_version=None,
ciphers=None, **kwargs):
"""
Initialize a Manager over TLS transport.
Parameters:
- host: str, hostname or IP address
- port: int, TLS port (default 6513)
- certfile: str, client certificate file
- keyfile: str, client private key file
- ca_certs: str, CA certificates file
- cert_reqs: int, certificate requirements
- ssl_version: int, SSL protocol version
- ciphers: str, cipher specification
Returns:
Manager: TLS-based NETCONF session manager
"""Unix Domain Socket Connection
Connects to NETCONF server via Unix domain socket (Unix platforms only).
def connect_uds(path, **kwargs):
"""
Initialize a Manager over Unix domain socket transport.
Parameters:
- path: str, path to Unix domain socket
Returns:
Manager: Unix socket-based NETCONF session manager
"""Call Home Connection
Handles NETCONF call-home connections where the device initiates the connection.
def call_home(host, port=4334, timeout=10, **kwargs):
"""
Handle NETCONF call-home connection.
Parameters:
- host: str, interface to bind and listen on
- port: int, port to listen on (default 4334)
- timeout: int, connection timeout in seconds
Returns:
Manager: Call-home NETCONF session manager
"""Device Handler Creation
Creates device-specific handlers for vendor optimizations and customizations.
def make_device_handler(device_params, ignore_errors=None):
"""
Create device handler for vendor-specific functionality.
Parameters:
- device_params: dict, device configuration including 'name' key
- ignore_errors: list, error patterns to ignore
Returns:
DeviceHandler: Device-specific handler instance
"""Manager Class
The Manager class provides the main interface for NETCONF operations and session management.
class Manager:
"""
Main NETCONF session manager providing operation dispatch and session control.
Supports context manager protocol for automatic session cleanup.
"""
def __init__(self, session, device_handler, timeout=30,
raise_mode=operations.RaiseMode.ALL):
"""
Initialize Manager instance.
Parameters:
- session: transport session instance
- device_handler: device-specific handler
- timeout: default operation timeout in seconds
- raise_mode: error raising behavior
"""
def __enter__(self):
"""Context manager entry."""
def __exit__(self, *args):
"""Context manager exit with session cleanup."""
def locked(self, target):
"""
Create lock context manager for datastore.
Parameters:
- target: str, datastore name ('running', 'candidate', 'startup')
Returns:
LockContext: Context manager for datastore locking
"""
def take_notification(self, block=True, timeout=None):
"""
Retrieve notification from queue.
Parameters:
- block: bool, whether to block waiting for notification
- timeout: float, timeout in seconds
Returns:
Notification or None: Retrieved notification object
"""
@property
def client_capabilities(self):
"""Capabilities: Client NETCONF capabilities."""
@property
def server_capabilities(self):
"""Capabilities: Server NETCONF capabilities."""
@property
def session_id(self):
"""str: NETCONF session ID."""
@property
def connected(self):
"""bool: Connection status."""
@property
def timeout(self):
"""float: Default operation timeout."""
@timeout.setter
def timeout(self, value):
"""Set default operation timeout."""
@property
def async_mode(self):
"""bool: Asynchronous operation mode."""
@async_mode.setter
def async_mode(self, mode):
"""Set asynchronous operation mode."""
@property
def raise_mode(self):
"""RaiseMode: Error raising behavior."""
@raise_mode.setter
def raise_mode(self, mode):
"""Set error raising behavior."""
@property
def huge_tree(self):
"""bool: Large XML tree support."""
@huge_tree.setter
def huge_tree(self, enabled):
"""Enable/disable large XML tree support."""Usage Examples
Basic SSH Connection
from ncclient import manager
# Simple SSH connection
with manager.connect_ssh(
host='192.168.1.1',
username='admin',
password='admin',
hostkey_verify=False
) as m:
print(f"Connected to session: {m.session_id}")
print("Server capabilities:")
for cap in m.server_capabilities:
print(f" {cap}")Device-Specific Connection
from ncclient import manager
# Junos device with specific parameters
device_params = {
'name': 'junos',
'local': False
}
with manager.connect_ssh(
host='juniper-router.example.com',
username='netconf-user',
key_filename='/path/to/private/key',
device_params=device_params
) as m:
# Device-specific operations available
passConnection with Error Handling
from ncclient import manager
from ncclient.operations import RaiseMode
# Configure error handling
errors_params = {
'ignore_errors': ['warning'],
'raise_mode': RaiseMode.ERRORS
}
manager_params = {
'timeout': 60
}
with manager.connect_ssh(
host='device.example.com',
username='admin',
password='admin',
errors_params=errors_params,
manager_params=manager_params
) as m:
# Operations with custom error handling
passInstall with Tessl CLI
npx tessl i tessl/pypi-ncclient