Examples¶
This section provides examples and background information to assist developers to maximize their use of the pySROS libraries.
Note
There are multiple ways in Python to achieve the requirements of an application. Examples shown are suggestions only; actual usage may differ depending on supported functionality and user configuration.
pySROS data structures¶
Having made a connection to an SR OS device it is important to understand the data structures returned by the
pysros.management.Datastore.get() function and those that can be sent to a router using
pysros.management.Datastore.set().
As described in the The pySROS data model section, data obtained from and sent to an SR OS device are described in a format referred to as a pySROS data structure.
This section provides some example outputs of the data structures for the respective YANG constructs.
YANG leaf structure¶
A YANG leaf is wrapped in the pysros.wrappers.Leaf class. The following example obtains a YANG leaf from
SR OS using the pysros.management.Datastore.get() method.
Obtaining the YANG leaf data structure¶
>>> from pysros.management import connect
>>> connection_object = connect()
>>> pysros_ds = connection_object.running.get("/nokia-conf:configure/system/name")
>>> pysros_ds
Leaf('sros')
To obtain the data only (excluding the wrapper), call .data on the obtained object.
Obtaining the data from a YANG leaf data structure¶
>>> pysros_ds.data
'sros'
YANG container structure¶
A YANG container is wrapped in the pysros.wrappers.Container class. The following example obtains a
YANG container from SR OS using the pysros.management.Datastore.get() method.
Obtaining the data from a YANG container data structure¶
>>> from pysros.management import connect
>>> connection_object = connect()
>>> data = connection_object.running.get('/nokia-state:state/router[router-name="Base"]/sfm-overload')
>>> data
Container({'state': Leaf('normal'), 'start': Leaf('2021-06-21T19:35:17.3Z'), 'time': Leaf(0)})
Note
State information is obtained by targeting the running datastore
YANG leaf-list structure¶
A YANG leaf-list is wrapped in the pysros.wrappers.LeafList class. The following example obtains a
YANG container from SR OS using the pysros.management.Datastore.get() method.
Obtaining the data from a YANG leaf-list data structure¶
>>> from pysros.management import connect
>>> connection_object = connect()
>>> data = connection_object.running.get('/nokia-conf:configure/router[router-name="Base"]/bgp/neighbor[ip-address="192.168.100.2"]/import/policy')
>>> data
LeafList(['demo', 'example-policy-statement'])
Note
The order of the entries in a leaf-list is important. Lists in Python are order aware.
YANG list structure¶
A YANG list is represented as a Python dict, as noted in the The pySROS data model section.
YANG lists define three elements:
a list name
a list key
a set of key-values
The YANG module example.yang shown below, demonstrates this:
example.yang¶
module example {
prefix "example";
namespace "urn:nokia.com:yang:example";
list listname {
key listkey;
leaf listkey {
type string;
}
}
}
The key of the YANG list (listkey) is a child element of the list (listname)
itself. Without knowledge of the structure of the specific YANG list, a developer would need to have an external
reference to be able to identify the key of the list to iterate through it or to reference specific items.
For this reason, the pySROS libraries translate YANG lists into Python dictionaries keyed on the values of the list
key (listkey) as opposed to Python lists.
The element /nokia-conf:configure/log/log-id is a YANG list in the nokia-conf YANG module. The YANG list name
is log-id and the YANG list’s key is name.
Note
The key of a YANG list can be identified using context-sensitive help within SR OS, from the YANG module or by
obtaining the pwc json-instance-path of an element in the YANG list, for example,
/nokia-conf:configure/log/log-id[name="10"].
Consider the following router configuration:
SR OS example log configuration¶
/configure log log-id "10" { }
/configure log log-id "10" { description "Log ten" }
/configure log log-id "11" { }
/configure log log-id "11" { description "Log eleven" }
If this list is obtained using pysros.Datastore.get() for the list log-id, the resultant data structure
is a Python dictionary, keyed on the values of the log-id list’s key, name:
Obtaining the data from a YANG list data structure¶
>>> from pysros.management import connect
>>> connection_object = connect()
>>> data = connection_object.running.get('/nokia-conf:configure/log/log-id')
>>> data
{'10': Container({'description': Leaf('Log ten'), 'name': Leaf('10')}),
'11': Container({'description': Leaf('Log eleven'), 'name': Leaf('11')})}
Configuring YANG lists¶
YANG lists may be configured using the pySROS libraries in many ways.
Assume the target configuration of the log-id list on SR OS is in the SR OS example log configuration as shown in the preceding example.
This is a YANG list named log-id, with one key name (name) and two key-values, 10 and 11.
Method 1¶
Configure the list with a payload that is a Python dictionary keyed on the YANG list key-value:
Configure YANG list using Python dict payload without key name¶
>>> connection_object = connect()
>>> path = '/nokia-conf:configure/log/log-id'
>>> payload = {'10': {'description': 'Log ten'}, '11': {'description': 'Log eleven'}}
>>> connection_object.candidate.set(path, payload)
Method 2¶
Configure the specific items in the list with a payload that is a Python dictionary containing the contents
of that list item without the key name and key-value information (The list’s key name and key-value are supplied
in the path).
Configure each YANG list entry in turn providing the key and key-value in the path¶
>>> connection_object = connect()
>>> list_entries = [("10", "Log ten"), ("11", "Log eleven")]
>>> for item in list_entries:
>>> payload = {'description': item[1]}
>>> path = '/nokia-conf:configure/log/log-id[name=' + item[0] + ']'
>>> connection_object.candidate.set(path, payload)
Method 3¶
Configure the specific items in the list with a payload that is a Python dictionary containing the contents
of that list item with the key name and key-value information provided (The list’s key name and key-value are
also supplied in the path). In this case, the contents of the payload dictionary for key name and key-value
must match the data provided in the path.
Configure each YANG list entry in turn providing the key and key-value in the path and payload¶
>>> connection_object = connect()
>>> list_entries = [("10", "Log ten"), ("11", "Log eleven")]
>>> for item in list_entries:
>>> payload = {'name': item[0], 'description': item[1]}
>>> path = '/nokia-conf:configure/log/log-id[name=' + item[0] + ']'
>>> connection_object.candidate.set(path, payload)
These examples are available in a single Python file here
YANG user-ordered list structure¶
A user-ordered list is similar to a YANG list structure.
Python dictionaries are not order aware. This is not important for most configuration items in the SR OS model-driven interfaces as SR OS ensures the required order is identified and achieved automatically. However, some specific lists in SR OS are order dependent, such as policies.
User-ordered lists are therefore represented as ordered dictionaries in Python.
Obtaining the data from a user-ordered YANG list data structure¶
>>> from pysros.management import connect
>>> connection_object = connect()
>>> data = connection_object.running.get('/nokia-conf:configure/policy-options/policy-statement[name="example-policy-statement"]/named-entry')
>>> data
OrderedDict({'one': Container({'entry-name': Leaf('one'), 'action': Container({'action-type': Leaf('accept')})}),
'three': Container({'entry-name': Leaf('three'), 'action': Container({'action-type': Leaf('accept')})})})
Connecting to the SR OS model-driven interface¶
Connecting when executing on a remote workstation¶
To access the model-driven interface of SR OS when executing a Python application from a remote workstation, use
the pysros.management.connect() method. Examples are shown in the method documentation.
Connecting when executing on SR OS¶
To access the model-driven interface of SR OS when executing a Python application on SR OS, use the
pysros.management.connect() method. The arguments to this method are ignored when executing on SR OS and
therefore the following two examples perform the same operation:
Connecting to the model-driven interfaces when executing on SR OS - example 1¶
from pysros.management import connect
connection_object = connect()
Connecting to the model-driven interfaces when executing on SR OS - example 2¶
from pysros.management import connect
connection_object = connect(host="192.168.1.1",
username="myusername",
password="mypassword")
Note
The preceding examples show the minimum Python code required to connect to the SR OS model-driven interfaces of
a device.
See the Handling exceptions section to ensure potential error scenarios in the
connection process are handled cleanly in your application and see the pysros.management.connect()
documentation for more detailed examples.
Handling exceptions¶
Python provides the ability to handle error situations through the use of exceptions. The pySROS libraries
provide a number of exceptions that are detailed within the pysros module documentation.
The following is an example of how to use these exceptions to handle specific error scenarios:
make_connection.py¶
#!/usr/bin/env python3
### make_connection.py
# Copyright 2021 Nokia
###
"""Example to show how to make a connection and handle exceptions"""
# Import sys for returning specific exit codes
import sys
# Import the connect method from the management pySROS sub-module
from pysros.management import connect
# Import the exceptions that are referenced so they can be caught on error.
from pysros.exceptions import ModelProcessingError
def get_connection(host=None, credentials=None):
"""Function definition to obtain a Connection object to a specific SR OS device
and access the model-driven information."""
# The try statement coupled with the except statements allow an operation(s) to be
# attempted and specific error conditions handled gracefully
try:
connection_object = connect(
host=host,
username=credentials["username"],
password=credentials["password"],
)
# Confirm to the user that the connection establishment completed successfully
print("Connection established successfully")
# Return the Connection object that we created
return connection_object
# This first exception is described in the pysros.management.connect method
# and references errors that occur during the creation of the Connection object.
# If the provided exception is raised during the execution of the connect method
# the information provided in that exception is loaded into the e1 variable for use
except RuntimeError as error1:
print(
"Failed to connect during the creation of the Connection object. Error:",
error1,
)
sys.exit(101)
# This second exception is described in the pysros.management.connect method
# and references errors that occur whilst compiling the YANG modules that have been
# obtained into a model-driven schema.
# If the provided exception is raised during the execution of the connect method the
# information provided in that exception is loaded into the e2 variable for use.
except ModelProcessingError as error2:
print("Failed to create model-driven schema. Error:", error2)
sys.exit(102)
# This last exception is a general exception provided in Python
# If any other unhandled specific exception is thrown the information provided in
# that exception is loaded into the e3 variable for use
except Exception as error3: # pylint: disable=broad-except
print("Failed to connect. Error:", error3)
sys.exit(103)
def main():
"""Example general/main function"""
# Define some user credentials to pass to the get_connect function
credentials = {"username": "myusername", "password": "mypassword"}
# Call the get_connection function providing a hostname/IP and the credentials
# Returns a Connection object for use in obtaining data from the SR OS device
# or configuring that device
connection_object = get_connection( # pylint: disable=unused-variable
host="192.168.1.1", credentials=credentials
)
assert connection_object
if __name__ == "__main__":
main()
Obtaining data and formatted output¶
This section provides some examples of obtaining data from SR OS and printing that data in various formats.
Show SDP state and descriptions¶
This examples creates a new show command that displays a list of the SDPs and their ID, description, administrative state, operational state, and far-end IP address.
This example demonstrates how to obtain data from various locations in the configure and state tree structure of
SR OS, how to manipulate and correlate this data, and how to use the pysros.pprint.Table class to
create SR OS style table output.
show_sdp_with_description.py¶
#!/usr/bin/env python3
### show_sdp_with_description.py
# Copyright 2021 Nokia
###
"""Example to show all SDPs with description"""
# Import the required libraries for the application.
import sys
from pysros.management import connect
from pysros.pprint import Table # pylint: disable=no-name-in-module
# Import the exceptions so they can be caught on error.
from pysros.exceptions import ModelProcessingError
# Global credentials dictionary for the purposes of this example. Global variables
# discouraged in operational applications.
credentials = {
"host": "192.168.1.1",
"username": "myusername",
"password": "mypassword",
"port": 830,
}
def get_connection(creds):
"""Function definition to obtain a Connection object to a specific SR OS device
and access the model-driven information."""
try:
connection_object = connect(
host=creds["host"],
username=creds["username"],
password=creds["password"],
port=creds["port"],
)
return connection_object
except RuntimeError as error1:
print(
"Failed to connect during the creation of the Connection object. Error:",
error1,
)
sys.exit(101)
except ModelProcessingError as error2:
print("Failed to create model-driven schema. Error:", error2)
sys.exit(102)
except Exception as error3: # pylint: disable=broad-except
print("Failed to connect:", error3)
sys.exit(103)
# Fuction definition to output a SR OS style table to the screen
def print_table(rows):
"""Setup and print the SR OS style table"""
# Define the columns that will be used in the table. Each list item
# is a tuple of (column width, heading).
cols = [
(10, "ID"),
(20, "Description"),
(10, "Adm"),
(10, "Opr"),
(15, "Far End"),
]
# Initalize the Table object with the heading and columns.
table = Table("Service Destination Points with Descriptions", cols, showCount="SDP")
# Print the output passing the data for the rows as an argument to the function.
table.print(rows)
# The main function definition
def main():
"""Main function to display all SDPs on an SR OS device"""
# Connect to the router.
connection_object = get_connection(credentials)
# Initialize the 'sdp_info' list that will be used to store the
# data obtained that we wish to use in the output.
sdp_info = []
# Obtain the SDP configuration information from the SR OS device. Ensure all
# default values are returned in addition to specifically set values.
sdp_conf = connection_object.running.get(
"/nokia-conf:configure/service/sdp", defaults=True
)
# Obtain the SDP state information from the SR OS device. Ensure all
# default values are returned in addition to specifically set values.
sdp_state = connection_object.running.get(
"/nokia-state:state/service/sdp", defaults=True
)
# Identify the SDP ID numbers and store this value as the variable 'id' and perform
# the following operations for every SDP.
for ident in sdp_conf.keys():
# Initalize the description variable as it is referenced later
description = None
# If the description of the SDP has been configured place the obtained
# description into the 'description' variable.
if "description" in sdp_conf[ident].keys():
description = sdp_conf[ident]["description"].data
# Store the administrative state of the SDP from the obtained
# configuration data.
admin_state = sdp_conf[ident]["admin-state"].data
# Store the far-end IP address of the SDP from the obtained
# configuration data.
far_end = sdp_conf[ident]["far-end"]["ip-address"].data
# Store the operational state of the SDP from the obtained
# state data.
oper_state = sdp_state[ident]["sdp-oper-state"].data
# Add the collected data to the 'sdp_info' list that will be used as
# the rows in the tabulated output.
sdp_info.append([ident, description, admin_state, oper_state, far_end])
# Print the table using the defined print_table function.
print_table(sdp_info)
# Disconnect from the model-driven interfaces of the SR OS node.
connection_object.disconnect()
# Returning 0 should be considered the same as completing the function with a
# thumbs up!
return 0
if __name__ == "__main__":
main()
The example output for this application is shown here:
===============================================================================
Service Destination Points with Descriptions
===============================================================================
ID Description Adm Opr Far End
-------------------------------------------------------------------------------
44 None enable down 192.168.100.2
-------------------------------------------------------------------------------
No. of SDP: 1
===============================================================================
Efficient YANG list key handling¶
There are occasions where a set of specific entries in a YANG list are required. In these situations, use one of the following to obtain the key values for the list.
Obtain the list from the node which is stored as a Python dict and then use
.keys()on the dict to obtain the key values.Use the
pysros.management.Datastore.get_list_keys()function to obtain the list of the key values without obtaining the full data structure of the YANG list.
Using the pysros.management. Datastore.get_list_keys() function is significantly
faster and uses less memory.
The following example compares and contrasts the different methods.
get_list_keys_usage.py¶
#!/usr/bin/env python3
### get_list_keys_usage.py
# Copyright 2022 Nokia
###
"""
Tested on: SR OS 22.2.R1
This example demonstrates the use of the get_list_keys function introduced in
release 22.2.1 of pySROS.
"""
# Import the time module to provide performance timers
import time
# Import the connect and sros methods from the management pySROS submodule
from pysros.management import connect, sros
def get_connection():
"""Function definition to obtain a Connection object to a specific SR OS device
and access model-driven information"""
# Use the sros() function to determine if the application is executed
# locally on an SR OS device, or remotely so that the same application
# can be developed to run locally or remotely. If the application is
# executed locally, call connect() and return the Connection object.
# If the application is executed remotely, the username and host is
# required as arguments, and a prompt for the password is displayed
# before calling connect(). Connection error handling is also checked.
# If the application is executed locally
if sros():
connection_object = connect() # pylint: disable=missing-kwoa
# Else if the application is executed remotely
else:
# Import sys for returning specific exit codes
import sys # pylint: disable=import-outside-toplevel
# Import getpass to read the password
import getpass # pylint: disable=import-outside-toplevel
# Import the exceptions so they can be caught on error
# pylint: disable=import-outside-toplevel
from pysros.exceptions import (
ModelProcessingError,
)
# Make sure we have the right number of arguments, the host can
# be an IP address or a hostname
if len(sys.argv) != 2:
print("Usage:", sys.argv[0], "username@host")
sys.exit(2)
# Split the username and host arguments
username_host = sys.argv[1].split("@")
if len(username_host) != 2:
print("Usage:", sys.argv[0], "username@host")
sys.exit(2)
# Get the password
password = getpass.getpass()
# The try statement coupled with the except statements allow an
# operation(s) to be attempted and specific error conditions handled
# gracefully
try:
connection_object = connect(
username=username_host[0],
host=username_host[1],
password=password,
)
return connection_object
# This first exception is described in the pysros.management.connect
# method and references errors that occur during the creation of the
# Connection object. If the provided exception is raised during
# the execution of the connect method the information provided in
# that exception is loaded into the runtime_error variable for use.
except RuntimeError as runtime_error:
print(
"Failed to connect during the creation of the Connection object."
)
print("Error:", runtime_error, end="")
print(".")
sys.exit(100)
# This second exception is described in the pysros.management.connect
# method and references errors that occur whilst compiling the YANG
# modules that have been obtained into a model-driven schema. If the
# provided exception is raised during the execution of the connect
# method the information provided in that exception is loaded into
# the model_proc_error variable for use.
except ModelProcessingError as model_proc_error:
print("Failed to compile YANG modules.")
print("Error:", model_proc_error, end="")
print(".")
sys.exit(101)
return connection_object
def get_list_keys_example_without_defaults(connection_object, path):
"""Function to obtain the keys of a YANG list (excluding default entries) using the
get_list_keys function.
"""
return connection_object.running.get_list_keys(path)
def get_list_keys_example_with_defaults(connection_object, path):
"""Function to obtain the keys of a YANG list (including default entries) using the
get_list_keys function.
"""
return connection_object.running.get_list_keys(path, defaults=True)
def get_list_then_extract_keys_example_without_defaults(
connection_object, path
):
"""Function to obtain the keys of a YANG list (excluding default entries) using the
get function and then calling keys() against the resulting data structure.
"""
return connection_object.running.get(path).keys()
def get_list_then_extract_keys_example_with_defaults(connection_object, path):
"""Function to obtain the keys of a YANG list (including default entries) using the
get function and then calling keys() against the resulting data structure.
"""
return connection_object.running.get(path, defaults=True).keys()
def compare_and_contrast():
"""Compare and contrast the different methods to obtain a list of key values
from a YANG list.
"""
# Obtain the Connection object for the device
connection_object = get_connection()
# The example path used in the example. This is the path to a YANG list.
path = "/nokia-conf:configure/router"
# Use get to obtain the list then select the keys from the resulting dict without defaults
starttime = time.perf_counter()
output = get_list_then_extract_keys_example_without_defaults(
connection_object, path
)
duration = round(time.perf_counter() - starttime, 4)
print("get without defaults\n", "Output:", output, "Time:", duration)
# Use get to obtain the list then select the keys from the resulting dict with defaults
starttime = time.perf_counter()
output = get_list_then_extract_keys_example_with_defaults(
connection_object, path
)
duration = round(time.perf_counter() - starttime, 4)
print("get with defaults\n", "Output:", output, "Time:", duration)
# Use get_list_keys to obtain the list keys without defaults
starttime = time.perf_counter()
output = get_list_keys_example_without_defaults(connection_object, path)
duration = round(time.perf_counter() - starttime, 4)
print(
"get_list_keys without defaults\n",
"Output:",
output,
"Time:",
duration,
)
# Use get_list_keys to obtain the list keys without defaults
starttime = time.perf_counter()
output = get_list_keys_example_with_defaults(connection_object, path)
duration = round(time.perf_counter() - starttime, 4)
print(
"get_list_keys with defaults\n", "Output:", output, "Time:", duration
)
connection_object.disconnect()
if __name__ == "__main__":
compare_and_contrast()
The example output for this application is shown here:
get without defaults
Output: dict_keys(['Base']) Time: 0.1393
get with defaults
Output: dict_keys(['Base', 'management', 'vpls-management']) Time: 0.7754
get_list_keys without defaults
Output: ['Base'] Time: 0.0859
get_list_keys with defaults
Output: ['Base', 'management', 'vpls-management'] Time: 0.1171
Multi-device hardware inventory¶
This example is created to be executed on a remote workstation. It connects to the devices that are supplied on input and obtains the hardware inventory from the chassis and line cards that can be used with external systems, with the output in JSON format.
get_inventory_remotely.py¶
#!/usr/bin/env python3
### get_inventory_remotely.py
# Copyright 2021 Nokia
###
"""Example to show obtaining a hardware inventory of multiple devices"""
# Import the required libraries
import sys
import ipaddress
import json
from pysros.management import connect, sros
from pysros.exceptions import ModelProcessingError
# Global credentials dictionary for the purposes of this example. Global variables
# discouraged in operational applications.
credentials = {"username": "myusername", "password": "mypassword"}
def get_connection(creds, host):
"""Function definition to obtain a Connection object to a specific SR OS device
and access the model-driven information."""
try:
connection_object = connect(
host=host,
username=creds["username"],
password=creds["password"],
)
return connection_object
except RuntimeError as error1:
print(
"Failed to connect during the creation of the Connection object. Error:",
error1,
)
sys.exit(101)
except ModelProcessingError as error2:
print("Failed to create model-driven schema. Error:", error2)
sys.exit(103)
except Exception as error3: # pylint: disable=broad-except
print("Failed to connect:", error3)
sys.exit(103)
def get_input_args():
"""Function definition to check that input arguments exist and that they are in
the correct format for processing."""
# Initialize the 'args' list that will be used to store the validate arguments
# for use later in the application
args = []
# Confirm that arguments have been provided
if len(sys.argv) > 1:
try:
# For every argument provided, confirm that the argument is a valid IP address
for arg in sys.argv[1:]:
args.append(ipaddress.ip_address(arg))
# Return the list of valid arguments (IP addresses)
return args
# If the provided arguments are not valid IP addresses provide this feedback and exit
# the application.
except Exception as error: # pylint: disable=broad-except
print("Argument is not valid IP address: ", error)
sys.exit(-1)
# If no arguments have been provided then print the usage information to the screen and
# exit the application
else:
print("This application expects a whitespace separated list of IP addresses\n")
print("Usage:", sys.argv[0], "<IP address> [IP address] ...\n")
sys.exit(-1)
return -1
def get_hardware_data(struct):
"""Function definition to iterate through a provided data-structure looking for a
hardware-data container and then create a dictionary of the contents."""
# Initialize the empty dictionary in case there is no hardware-data container in
# the provided data-structure.
build_hash = {}
# Identify the hardware-data container in the data-structure and create the dictionary.
for k in struct["hardware-data"]:
build_hash[k] = struct["hardware-data"][k].data
# Return the dictionary of hardware-data inventory items.
return build_hash
def get_chassis_inventory(connection_object):
"""Function definition to obtain chassis state information from SR OS and obtain the
required data in order to build the inventory."""
# Obtain the data from the SR OS device. The data in this example is only obtained
# from the first chassis per host.
chassis_state = connection_object.running.get(
'/nokia-state:state/chassis[chassis-class="router"][chassis-number="1"]'
)
# Define some constants that describe the chassis itself
hwtype = "chassis"
number = 1
# Obtain the dictionary of the hardware-data elements and their values from the
# chassis_state data-structure.
build_hash = get_hardware_data(chassis_state)
# Initialize the results dictionary with the chassis information using the above constants
# and the obtained hardware-data.
hash_data = {hwtype: {number: build_hash}}
# For every element in the original chassis_state data-structure iterate through the
# child elements (depth 1) and select the fans and power-supplies.
for item in chassis_state.keys():
if item in ["fan", "power-supply"]:
item_num_hash = {}
# For every fan and power-supply obtain the hardware-data information and add
# it to the return dictionary.
for item_num in chassis_state[item]:
item_num_hash[item_num] = get_hardware_data(
chassis_state[item][item_num]
)
hash_data[item] = item_num_hash
# Return the dictionary containing all chassis, fan and power-supply hardware inventory data.
return hash_data
def get_card_inventory(connection_object):
"""Function definition to obtain card state information from SR OS and obtain the
required data in order to build the inventory."""
# Obtain the state data about line cards from the SR OS device.
card_state = connection_object.running.get("/nokia-state:state/card")
# Initialize the dictionary because no line cards may exist.
hash_data = {}
# For every line card installed obtain the hardware-data for the inventory.
for item_number in card_state.keys():
hash_data[item_number] = get_hardware_data(card_state[item_number])
# Return a dictionary of cards with the obtained data.
return {"card": hash_data}
def main():
"""The 'main' function definition. Check that the application is not being executed on SR OS
and then obtain the provided arguments and iterate through the hosts to obtain and build
a hardware inventory which is outputted in JSON format."""
# Check that the application is not being executed on SR OS.
# If it is, provide a warning and exit.
if sros():
print(
"This application cannot be run on the SR OS node. Please run it on an external device"
)
sys.exit(200)
else:
# Obtain the host IP addresses provided as input arguments.
hosts = get_input_args()
# Initialize the inventory dictionary to cover the case where no hardware data is returned.
chassis_inv = {}
# For every host connect to the model-driven interface and obtain the inventory/
for host in hosts:
# Obtain a Connection object for the host.
connection_object = get_connection(credentials, str(host))
# Obtain the inventory of chassis, fan and power-supply data.
per_host_inv = get_chassis_inventory(connection_object)
# Obtain the inventory of line card data.
per_host_inv.update(get_card_inventory(connection_object))
# Link the completed inventory with the specific host in the output dictionary.
chassis_inv[str(host)] = per_host_inv
# Close the connection to the model-driven interface of the host.
connection_object.disconnect()
# Output the obtained inventory in indented JSON format.
print(json.dumps(chassis_inv, indent=4))
if __name__ == "__main__":
main()
Local language output¶
This example demonstrates the ability to display unicode (UTF-8) characters on SR OS. This allows for the addition of local language personalization for developers.
local_language_output.py¶
#!/usr/bin/env python3
### local_language_output.py
# Copyright 2021 Nokia
###
"""Example to UTF-8 output - execute this example on SR OS"""
def main():
"""The main function definition"""
# Print local language example output as UTF-8 characters can be output on SR OS
# using the Python 3
print(
"pySROS knižnica umožňuje vývoj príkazov pre zobrazenie stavových "
"premenných v lokálnom jazyku vďaka podpore UTF-8 znakov."
)
if __name__ == "__main__":
main()
Filesystem access¶
Filesystem access is provided to the local SR OS filesystem using the standard Python 3 methods.
Specific adaptations have been provided for some libraries. See uio, uos
and uos.path for more information.
Note
Filesystem access is provided using the SR OS profile of the executing user and respects any permissions or restrictions therein.
Important
Python applications triggered from EHS and CRON have system access to the filesystem.
The ability to read and write to the filesystem provides many possibilities to the developer, including the ability to maintain a persistent state between executions. This enables a developer to choose to evaluate something based on the last time the application was run, in addition to the instantaneous data available.
filesystem_example.py¶
#!/usr/bin/env python3
### filesystem_example.py
# Copyright 2022 Nokia
###
"""Example to demonstrate local filesystem access on an SR OS device.
Tested on: SR OS 22.7.R1
"""
# pylint: disable=broad-except, eval-used, unspecified-encoding
import sys
from pysros.management import connect, sros
def update_file(filename=None, operation="get", counter=None, stat_value=None):
"""Read and write from the file on the SR OS filesystem.
:param filename: Full path to the filename to read/write to/from.
:type filename: str
:param operation: Define whether the action is to ``get`` (read) from the
file or ``set`` (write) to the file.
:type operation: str
:param counter: Counter variable for number of times the file has been
written to.
:type counter: int
:param stat_value: Value of the statistic being recorded to the file.
:type stat_value: int
:returns: Returns the counter and statistic
:rtype: tuple
"""
if operation == "get":
try:
with open(filename, "r") as file_object:
# Read the tuple in from the file, if it does not exist, set it to (0, stat_value)
# where stat_value where stat_value exists (the program will exit earlier if the
# statistic cannot be obtained.
return_tuple = eval(file_object.read() or (0, stat_value))
return return_tuple
except OSError:
# If the file cannot be read (for example, because it does not exist) then return
# the default counter 0 and the current statistic value
return (0, stat_value)
except Exception as error:
print("Failed:", error)
sys.exit(6)
elif operation == "set":
try:
with open(filename, "w") as file_object:
# Write to the file the current counter and stat_value as a tuple
file_object.write(str((counter, stat_value)))
return 0
except Exception as error:
print("Failed:", error)
sys.exit(7)
else:
sys.exit(3)
def get_stat_value(connection_object):
"""Obtain the statistics from the node. The statistic gathered in
this example is the number of received octets from a specific BGP peer
at the neighbor IP address of 192.168.10.2. You can replace this path with
a path of your choosing.
:param connection_object: Connection object referencing the router connection.
:type connection_object: :py:class:`pysros.Connection`
:returns: pySROS dataset containing the requested statistics
:rtype: dict
"""
return connection_object.running.get(
# pylint: disable=line-too-long
'/nokia-state:state/router[router-name="Base"]/bgp/neighbor[ip-address="192.168.10.2"]/statistics/received/octets'
)
def main():
"""This is the main function, the example code begins here."""
# Obtain a connection to the router. The example uses connect() without
# credentials as this example is designed to operate on the SR OS device.
try:
connection_object = connect() # pylint: disable=missing-kwoa
except Exception as error:
print("Failed to obtain a connection:", error)
sys.exit(5)
# Attempt to obtain the statistical data from the node.
try:
stat_value = get_stat_value(connection_object).data
except Exception as error:
print("Failed to obtain statistics\n", error)
sys.exit(2)
# Provide the full path to the file the will be read from and written to.
filename = "cf3:\\counter.txt"
# Obtain the data the file
mytuple = update_file(filename=filename, stat_value=stat_value)
# Output to the screen the details of this run and the difference between this
# run the and last.
print("This command has been run", mytuple[0] + 1, "times")
print(
"Number of received octets for BGP peer 192.168.10.2 (last run/this run):",
mytuple[1],
"/",
stat_value,
)
print(
"The difference between the last run and this run is:",
stat_value - mytuple[1],
)
# Increase the counter and update the file with the new data.
counter = mytuple[0] + 1
update_file(
filename=filename,
operation="set",
counter=counter,
stat_value=stat_value,
)
if __name__ == "__main__":
# Check that the application is being run on an SR OS device, if not then
# exit.
if sros():
main()
else:
print("This example is designed to operate on an SR OS node.")
sys.exit(10)
The example output of this application is shown below.
[/]
A:myusername@sros# pyexec filesystem_example.py
This command has been run 20 times
Number of received octets for BGP peer 192.168.100.2 (last run/this run): 205754 / 209022
The difference between the last run and this run is: 3268
[/]
A:myusername@sros# pyexec filesystem_example.py
This command has been run 21 times
Number of received octets for BGP peer 192.168.100.2 (last run/this run): 209022 / 209022
The difference between the last run and this run is: 0
[/]
A:myusername@sros# pyexec filesystem_example.py
This command has been run 22 times
Number of received octets for BGP peer 192.168.100.2 (last run/this run): 209022 / 209041
The difference between the last run and this run is: 19
Converting data formats¶
The pysros.management.Connection.convert() method converts
various YANG-modeled data formats to assist with input and output
data manipulation when integrating with pySROS and external systems.
The supported formats are:
pySROS data structure format (pysros)
XML (xml)
JSON IETF according to RFC 7951 (json)
Detailed documentation of the convert method is located in the
pysros.management.Connection.convert() section.
The following example code provides a class that generates an example
json-instance-path and associated example data in a chosen format that
can be used in the pysros.management.Connection.convert() method.
It then performs the following conversions using a connection to a node and outputs the results:
Input |
Output |
|---|---|
pysros |
pysros |
pysros |
xml |
pysros |
json |
xml |
xml |
xml |
pysros |
xml |
json |
json |
json |
json |
pysros |
json |
xml |
The example code can be found in the examples directory and is shown below:
Example showing the convert method for various formats¶
#!/usr/bin/env python3
### convert_example.py
# Copyright 2022 Nokia
###
"""Example to demonstrate the convert function to manipulate data formats.
Tested on: SR OS 22.10.R1
"""
# Import sys library
import sys
# Import the connect method from the management pySROS sub-module
from pysros.management import connect
# Import the exceptions that are referenced so they can be caught on error.
from pysros.exceptions import ModelProcessingError
class Data: # pylint: disable=too-few-public-methods
"""Create an object containing the input data in the required format
and the path used in the convert method.
:parameter input_format: The type of data to generate, either:
``pysros``, ``xml`` or ``json``.
:type input_format: str
:returns: Object with ``path`` and ``payload`` parameters in the desire format.
:rtype: :py:class:`Data`
:raises Exception: Raises a general Exception if the ``input_format`` is invalid.
"""
def __init__(self, input_format):
self.path = "/nokia-conf:configure/system/management-interface"
if input_format == "pysros":
self.payload = self._gen_pysros()
elif input_format == "xml":
self.payload = self._gen_xml()
elif input_format == "json":
self.payload = self._gen_json()
else:
raise ValueError("Valid data formats are pysros, xml or json")
def _gen_pysros(self):
"""Private method to generate simple pySROS formatted example data
without namespaces.
"""
return {
"snmp": {"admin-state": "disable"},
"netconf": {"admin-state": "enable", "auto-config-save": True},
"yang-modules": {
"nmda": {"nmda-support": True},
"openconfig-modules": True,
},
"cli": {"md-cli": {"auto-config-save": True}},
"configuration-mode": "model-driven",
}
def _gen_xml(self):
"""Private method to generate simple XML formatted example data
without namespaces.
"""
example_data = """
<cli>
<md-cli>
<auto-config-save>true</auto-config-save>
</md-cli>
</cli>
<netconf>
<admin-state>enable</admin-state>
<auto-config-save>true</auto-config-save>
</netconf>
<yang-modules>
<openconfig-modules>true</openconfig-modules>
<nmda>
<nmda-support>true</nmda-support>
</nmda>
</yang-modules>
<snmp>
<admin-state>disable</admin-state>
</snmp>
"""
return example_data
def _gen_json(self):
"""Private method to generate simple JSON IETF formatted example data
without namespaces.
"""
example_data = """
{
"configuration-mode": "model-driven",
"cli": {
"md-cli": {
"auto-config-save": true
}
},
"netconf": {
"admin-state": "enable",
"auto-config-save": true
},
"yang-modules": {
"openconfig-modules": true,
"nmda": {
"nmda-support": true
}
},
"snmp": {
"admin-state": "disable"
}
}
"""
return example_data
def __str__(self):
"""Allow the object to be printed directly."""
return "Path: " + self.path + "\n" + "Payload: " + self.payload
def get_connection(host=None, credentials=None, port=830):
"""Function definition to obtain a Connection object to a specific SR OS device
and access the model-driven information.
:parameter host: The hostname or IP address of the SR OS node.
:type host: str
:paramater credentials: The username and password to connect
to the SR OS node.
:type credentials: dict
:parameter port: The TCP port for the connection to the SR OS node.
:type port: int
:returns: Connection object for the SR OS node.
:rtype: :py:class:`pysros.management.Connection`
"""
# The try statement coupled with the except statements allow an operation
# attempt and specific error conditions handled gracefully
try:
print("-" * 79)
print("Obtaining connection to", host)
connection_object = connect(
host=host,
username=credentials["username"],
password=credentials["password"],
port=port,
hostkey_verify=False,
)
# Confirm to the user that the connection establishment completed successfully
print("Connection established successfully")
# Return the created Connection object
return connection_object
# This first exception is described in the pysros.management.connect method
# and references errors that occur during the creation of the Connection object.
# If the provided exception is raised during the execution of the connect method,
# the information provided in that exception is loaded into the error1 variable for use
except RuntimeError as error1:
print(
"Failed to connect during the creation of the Connection object. Error:",
error1,
)
sys.exit(101)
# This second exception is described in the pysros.management.connect method
# and references errors that occur when compiling the YANG modules that have been
# obtained into a model-driven schema.
# If the provided exception is raised during the execution of the connect method, the
# information provided in that exception is loaded into the error2 variable for use.
except ModelProcessingError as error2:
print("Failed to create model-driven schema. Error:", error2)
sys.exit(102)
# This last exception is a general exception provided in Python
# If any other unhandled specific exception occurs, the information provided in
# that exception is loaded into the error3 variable for use
except Exception as error3: # pylint: disable=broad-except
print("Failed to connect. Error:", error3)
sys.exit(103)
def converting(input_format, output_format, data, connection_object):
"""Perform the conversion from a given format to another format.
:parameter input_format: The type of data used on input, either:
``pysros``, ``xml`` or ``json``.
:type input_format: str
:parameter output_format: The type of data used on output, either:
``pysros``, ``xml`` or ``json``.
:type input_format: str
:parameter data: The example data object.
:type data: :py:class:`Data`
:parameter connection_object: The connection object for a specific node.
:type connection_object: :py:class:`pysros.management.Connection`
"""
print("-" * 79)
print("Converting", input_format, "to", output_format, "\n")
print("The path used as the YANG modelled root for the data is:")
print(data.path, "\n")
print("The payload is:")
print(data.payload, "\n")
print("The converted result is:")
print(
connection_object.convert(
path=data.path,
payload=data.payload,
source_format=input_format,
destination_format=output_format,
pretty_print=True,
)
)
def main():
"""The main procedure. The execution starts here."""
connection_object = get_connection(
"192.168.1.1", {"username": "admin", "password": "admin"}
)
data = Data("pysros")
converting("pysros", "pysros", data, connection_object)
data = Data("pysros")
converting("pysros", "xml", data, connection_object)
data = Data("pysros")
converting("pysros", "json", data, connection_object)
data = Data("xml")
converting("xml", "xml", data, connection_object)
data = Data("xml")
converting("xml", "pysros", data, connection_object)
data = Data("xml")
converting("xml", "json", data, connection_object)
data = Data("json")
converting("json", "json", data, connection_object)
data = Data("json")
converting("json", "pysros", data, connection_object)
data = Data("json")
converting("json", "xml", data, connection_object)
if __name__ == "__main__":
main()
The example output of this application is shown below.
-------------------------------------------------------------------------------
Obtaining connection to 192.168.1.1
Connection established successfully
-------------------------------------------------------------------------------
Converting pysros to pysros
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
{'snmp': {'admin-state': 'disable'}, 'netconf': {'admin-state': 'enable', 'auto-config-save': True}, 'yang-modules': {'nmda': {'nmda-support': True}, 'openconfig-modules': True}, 'cli': {'md-cli': {'auto-config-save': True}}, 'configuration-mode': 'model-driven'}
The converted result is:
{'snmp': Container({'admin-state': Leaf('disable')}), 'netconf': Container({'admin-state': Leaf('enable'), 'auto-config-save': Leaf(True)}), 'yang-modules': Container({'nmda': Container({'nmda-support': Leaf(True)}), 'openconfig-modules': Leaf(True)}), 'cli': Container({'md-cli': Container({'auto-config-save': Leaf(True)})}), 'configuration-mode': Leaf('model-driven')}
-------------------------------------------------------------------------------
Converting pysros to xml
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
{'snmp': {'admin-state': 'disable'}, 'netconf': {'admin-state': 'enable', 'auto-config-save': True}, 'yang-modules': {'nmda': {'nmda-support': True}, 'openconfig-modules': True}, 'cli': {'md-cli': {'auto-config-save': True}}, 'configuration-mode': 'model-driven'}
The converted result is:
<snmp xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<admin-state>disable</admin-state>
</snmp>
<netconf xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<admin-state>enable</admin-state>
<auto-config-save>true</auto-config-save>
</netconf>
<yang-modules xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<nmda>
<nmda-support>true</nmda-support>
</nmda>
<openconfig-modules>true</openconfig-modules>
</yang-modules>
<cli xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<md-cli>
<auto-config-save>true</auto-config-save>
</md-cli>
</cli>
<configuration-mode xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">model-driven</configuration-mode>
-------------------------------------------------------------------------------
Converting pysros to json
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
{'snmp': {'admin-state': 'disable'}, 'netconf': {'admin-state': 'enable', 'auto-config-save': True}, 'yang-modules': {'nmda': {'nmda-support': True}, 'openconfig-modules': True}, 'cli': {'md-cli': {'auto-config-save': True}}, 'configuration-mode': 'model-driven'}
The converted result is:
{
"nokia-conf:snmp": {
"admin-state": "disable"
},
"nokia-conf:netconf": {
"admin-state": "enable",
"auto-config-save": true
},
"nokia-conf:yang-modules": {
"nmda": {
"nmda-support": true
},
"openconfig-modules": true
},
"nokia-conf:cli": {
"md-cli": {
"auto-config-save": true
}
},
"nokia-conf:configuration-mode": "model-driven"
}
-------------------------------------------------------------------------------
Converting xml to xml
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
<cli>
<md-cli>
<auto-config-save>true</auto-config-save>
</md-cli>
</cli>
<netconf>
<admin-state>enable</admin-state>
<auto-config-save>true</auto-config-save>
</netconf>
<yang-modules>
<openconfig-modules>true</openconfig-modules>
<nmda>
<nmda-support>true</nmda-support>
</nmda>
</yang-modules>
<snmp>
<admin-state>disable</admin-state>
</snmp>
The converted result is:
<cli xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<md-cli>
<auto-config-save>true</auto-config-save>
</md-cli>
</cli>
<netconf xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<admin-state>enable</admin-state>
<auto-config-save>true</auto-config-save>
</netconf>
<yang-modules xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<openconfig-modules>true</openconfig-modules>
<nmda>
<nmda-support>true</nmda-support>
</nmda>
</yang-modules>
<snmp xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<admin-state>disable</admin-state>
</snmp>
-------------------------------------------------------------------------------
Converting xml to pysros
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
<cli>
<md-cli>
<auto-config-save>true</auto-config-save>
</md-cli>
</cli>
<netconf>
<admin-state>enable</admin-state>
<auto-config-save>true</auto-config-save>
</netconf>
<yang-modules>
<openconfig-modules>true</openconfig-modules>
<nmda>
<nmda-support>true</nmda-support>
</nmda>
</yang-modules>
<snmp>
<admin-state>disable</admin-state>
</snmp>
The converted result is:
{'cli': Container({'md-cli': Container({'auto-config-save': Leaf(True)})}), 'netconf': Container({'admin-state': Leaf('enable'), 'auto-config-save': Leaf(True)}), 'yang-modules': Container({'openconfig-modules': Leaf(True), 'nmda': Container({'nmda-support': Leaf(True)})}), 'snmp': Container({'admin-state': Leaf('disable')})}
-------------------------------------------------------------------------------
Converting xml to json
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
<cli>
<md-cli>
<auto-config-save>true</auto-config-save>
</md-cli>
</cli>
<netconf>
<admin-state>enable</admin-state>
<auto-config-save>true</auto-config-save>
</netconf>
<yang-modules>
<openconfig-modules>true</openconfig-modules>
<nmda>
<nmda-support>true</nmda-support>
</nmda>
</yang-modules>
<snmp>
<admin-state>disable</admin-state>
</snmp>
The converted result is:
{
"nokia-conf:cli": {
"md-cli": {
"auto-config-save": true
}
},
"nokia-conf:netconf": {
"admin-state": "enable",
"auto-config-save": true
},
"nokia-conf:yang-modules": {
"openconfig-modules": true,
"nmda": {
"nmda-support": true
}
},
"nokia-conf:snmp": {
"admin-state": "disable"
}
}
-------------------------------------------------------------------------------
Converting json to json
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
{
"configuration-mode": "model-driven",
"cli": {
"md-cli": {
"auto-config-save": true
}
},
"netconf": {
"admin-state": "enable",
"auto-config-save": true
},
"yang-modules": {
"openconfig-modules": true,
"nmda": {
"nmda-support": true
}
},
"snmp": {
"admin-state": "disable"
}
}
The converted result is:
{
"nokia-conf:configuration-mode": "model-driven",
"nokia-conf:cli": {
"md-cli": {
"auto-config-save": true
}
},
"nokia-conf:netconf": {
"admin-state": "enable",
"auto-config-save": true
},
"nokia-conf:yang-modules": {
"openconfig-modules": true,
"nmda": {
"nmda-support": true
}
},
"nokia-conf:snmp": {
"admin-state": "disable"
}
}
-------------------------------------------------------------------------------
Converting json to pysros
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
{
"configuration-mode": "model-driven",
"cli": {
"md-cli": {
"auto-config-save": true
}
},
"netconf": {
"admin-state": "enable",
"auto-config-save": true
},
"yang-modules": {
"openconfig-modules": true,
"nmda": {
"nmda-support": true
}
},
"snmp": {
"admin-state": "disable"
}
}
The converted result is:
{'configuration-mode': Leaf('model-driven'), 'cli': Container({'md-cli': Container({'auto-config-save': Leaf(True)})}), 'netconf': Container({'admin-state': Leaf('enable'), 'auto-config-save': Leaf(True)}), 'yang-modules': Container({'openconfig-modules': Leaf(True), 'nmda': Container({'nmda-support': Leaf(True)})}), 'snmp': Container({'admin-state': Leaf('disable')})}
-------------------------------------------------------------------------------
Converting json to xml
The path used as the YANG modelled root for the data is:
/nokia-conf:configure/system/management-interface
The payload is:
{
"configuration-mode": "model-driven",
"cli": {
"md-cli": {
"auto-config-save": true
}
},
"netconf": {
"admin-state": "enable",
"auto-config-save": true
},
"yang-modules": {
"openconfig-modules": true,
"nmda": {
"nmda-support": true
}
},
"snmp": {
"admin-state": "disable"
}
}
The converted result is:
<configuration-mode xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">model-driven</configuration-mode>
<cli xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<md-cli>
<auto-config-save>true</auto-config-save>
</md-cli>
</cli>
<netconf xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<admin-state>enable</admin-state>
<auto-config-save>true</auto-config-save>
</netconf>
<yang-modules xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<openconfig-modules>true</openconfig-modules>
<nmda>
<nmda-support>true</nmda-support>
</nmda>
</yang-modules>
<snmp xmlns="urn:nokia.com:sros:ns:yang:sr:conf" xmlns:nokia-attr="urn:nokia.com:sros:ns:yang:sr:attributes">
<admin-state>disable</admin-state>
</snmp>