Introduction
Thank you Kirill Pletnev for letting me know about ttp module.
ttp – Template Text Parser – is an easier to use parser than TextFSM, user does not need to know regex, the regex has already been defined, but how comprehensive is the regex is yet for me to discover. You can refer to the actual patterns from https://github.com/dmulyalin/ttp/tree/master/ttp/patterns. The regex patterns are not very complex, on ip addresses and mac addresses it is assumed that the appliance has already done the checks hence the patterns do not need to be comprehensive, however to test ip address using regex is not feasible as ip addresses are in bits, a better module to check ip address validity is always ipaddress which is a built-in module of python3, I have a post to check on ipv4 address validity here, the most complex bit calculation has already been made for you by the python developers. The reason why python is easier to learn is because it has a huge community with people who have time and do not ask for monetary benefits contributed to the extension/modules, even if I have made up an awesome python program it is not because i am smart is because there are people in the community who have made programming easier for me, so my part of the code also belongs to the community.
ttp template
access-list {{acl_name}} extended {{action}} {{protocol}} {{src_type}} {{src_object}} {{dst_type}} {{dst_object}} {{svc_obj_type}} {{service}} log
This is the output of show run access-list in Cisco ASA. As you can see I only need to define the variables (dictionary key names) and place them in the output the rest will be matched by ttp and map the value back to the key names and produce a list of dictionary. The output produces a nested list of dictionary hence in order to refer to the actual list of dictionary you need to get it from index 0 of the outer list.
Demonstration code on how to use ttp
from network.ciscoasa import connect_asa_host
from ttp import ttp
from nornir.plugins.tasks.networking import netmiko_send_command
"""
Easier to use text parser than TextFSM, see the examples here:
https://github.com/dmulyalin/ttp
User does not need to know regex you just need to place the variables within
the text, ttp will just map your variable with the information it matches.
Can get more information on the regex: https://github.com/dmulyalin/ttp/blob/master/ttp/patterns/get_pattern.py
It seems it is designed to match network parameters.
The example here will connect to the Cisco ASA - fw02, and get the output of show run access-list.
The result will then be parsed and produce a nested list of dictionary based on the ttp template.
to refer to the actual list of dictionary you need to refer by parser.result(format="json")[0]
"""
fw02 = connect_asa_host(hostname="fw02")
response = fw02.run(task=netmiko_send_command, command_string="sh run access-list")
with open("../network/templates/acl_match.ttp", "r") as t:
template = t.read()
result_for_parsing = response["fw02"][0].result
parser = ttp(data=result_for_parsing, template=template)
parser.parse()
print(parser.result(format="json")[0])
Demonstration
The list of access-list is huge, hence a portion is shown here.
The script output:
Not all are shown as it is extremely long, but if you match the template with the output you will see that ttp is matching the anchored value with the key name which I specified in the template.
Compare this template:
access-list {{acl_name}} extended {{action}} {{protocol}} {{src_type}} {{src_object}} {{dst_type}} {{dst_object}} {{svc_obj_type}} {{service}} log
with one of the dictionary:
{
"acl_name": "abcd",
"action": "permit",
"dst_object": "192.168.1.200",
"dst_type": "host",
"protocol": "tcp",
"service": "1433",
"src_object": "172.20.1.200",
"src_type": "host",
"svc_obj_type": "eq"
}
Because the output is immediately a dictionary it is easier for me to further process the output to the output response I desired, remember TextFSM is producing a nested list of all matched values, I need to design function to process the list into dictionary. TTP is very promising and suitable for people who do not want to learn too much regex, but still I would recommend people to learn regex since it is used everywhere to match interesting data we cannot rely too much on modules made by other people, if the script made by others failed we ourselves need to know how to fix it and not wait for people to fix for us, people in the community has already shared with us their products and they have their work and life it is good to report bugs but not good if we push the community to fix the problem of their module.
cyruslab 
Absolutely agree regarding learning regex and want to share an interesting resource https://ihateregex.io/ which shows what a particular expression match and how it does it.
Wow Kirill you are very resourceful… i did not know the web until now… it gave the actual regex to match the thing i need… I remember I took a lot of time to match multiple outputs of show access-list on cisco asa… there are many combinations.
Are you good in class? I found a solution to return values while running threads, but I got no idea what I was writing, the class looks like this:
from threading import Thread
“””
See reference: https://stackoverflow.com/questions/6893968/how-to-get-the-return-value-from-a-thread-in-python/40344234#40344234
Check the answer by GuySoft.
“””
class AsaThread(Thread):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._return = None
def run(self):
if self._target is not None:
self._return = self._target(*self._args, **self._kwargs)
def join(self, *args, **kwargs):
super().join(*args, **kwargs)
return self._return
I do know why super().__init__ is required, it is a requirement from the threading.Thread class, but what i did not know is these:
if self._target is not None:
self._return = self._target(*self._args, **self._kwargs)
I have not defined self_args and self_kwargs, but it worked…
I feel pretty confident with OOP but I’m not familiar with Tread class internals since it looks unnecessary complicated.
I prefer to do it like this;
I have Device class
class Device:
def __init__(self, nb_device: NetboxDevice) -> None:
self.device = nb_device
self._conn = None
def collect_data(self) -> None:
<do stuff: rest, netconf, whatever
nb_device contains Netbox data, so I can do nb.device.hostname to get a hostname
Then I have these functions
for threads
def othread(method, objects, *args, **kwargs):
output = []
with concurrent.futures.ThreadPoolExecutor(max_workers=256) as executor:
fs = []
for obj in objects:
fs.append(executor.submit(getattr(obj, method), *args, **kwargs))
for f in concurrent.futures.as_completed(fs):
try:
result = f.result()
except Exception:
traceback.print_exc(file=sys.stdout)
else:
output.append(result)
return output
and for processes
def oprocess(method, objects, *args, **kwargs):
output = []
with concurrent.futures.ProcessPoolExecutor(max_workers=256) as executor:
fs = []
for obj in objects:
fs.append(executor.submit(getattr(obj, method), *args, **kwargs))
for f in concurrent.futures.as_completed(fs):
try:
result = f.result()
except Exception:
traceback.print_exc(file=sys.stdout)
else:
output.append(result)
return output
So I have a list of my devices (instances of Device class)
devices = [, …]
and I run `collect_data` method on every one of them. Each is executed in a separate process
oprocess(‘collect_data’, devices)
You can do so `collect_data` complemented Device instance, for example, collect interface counters and make it and attribute, so you can do something like this:
for device in devices:
print(device.interface_data)
Or you can make `collect_data` to return some values:
results = oprocess(‘collect_data’, devices)
In this case `results` will be a list of `interface_data` of every device.
I think I know how it works… the original threading.Thread run and join methods do not return any value, hence all return values will be None, my class simply added / override the original class by simply return value from the task (function in threading context). I compare the original with my class, the parameters are all from the parent class.
The only var the parent class does not have is self._return, hence I pre-configured to None as a placeholder… i think this is rather smart….
I think I need some time on concurrent.futures, I used it once… but I forgot how to, i remember it was easier to use than the raw threading.Thread. Thanks 😀
I guess you already know that concurrent.futures just simply take over executing a function/method and return a result. No queues no worries =)