Examples

In examples are used configuration files. config.yaml:

device_list: credentials.yaml

credentials.yaml

- device_type: cisco_asa
  host: 1.1.1.1
  password: ****
  username: ****
- device_type: cisco_ios
  host: 2.2.2.2
  password: ****
  username: ****
- device_type: fujitsu_switch
  host: 3.3.3.3
  password: ****
  username: ****
- device_type: hp_comware
  host: 4.4.4.4
  password: ****
  username: ****
- device_type: hp_comware_limited
  host: 5.5.5.5
  password: ****
  username: ****
  cmdline_password: '512900'
- device_type: arista_eos
  host: 6.6.6.6
  password: ****
  username: ****
- device_type: juniper_junos
  host: 7.7.7.7
  password: ****
  username: ****

ASA example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as asa:
        # Testing sending simple command
        out = await asa.send_command('show run')
        print(out)
        # Testing interactive dialog
        out = await asa.send_command("copy r scp:", pattern=r'\[running-config\]\?', strip_command=False)
        out += await asa.send_command("\n", pattern=r'\[\]\?', strip_command=False)
        out += await asa.send_command("\n", strip_command=False)
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_asa']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

NX-OS example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as nxos:
        # Testing sending simple command
        out = await nxos.send_command('show run', strip_command=True)
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_nxos']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

IOS example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as ios:
        # Testing sending simple command
        out = await ios.send_command("show ver")
        print(out)
        # Testing sending configuration set
        commands = ["line console 0", "exit"]
        out = await ios.send_config_set(commands)
        print(out)
        # Testing sending simple command with long output
        out = await ios.send_command("show run")
        print(out)
        # Testing interactive dialog
        out = await ios.send_command("conf", pattern=r'\[terminal\]\?', strip_command=False)
        out += await ios.send_command("term", strip_command=False)
        out += await ios.send_command("exit", strip_command=False, strip_prompt=False)
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_ios']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

IOS XR example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as iosxr:
        # Testing sending simple command
        out = await iosxr.send_command("show ver")
        print(out)
        # Testing sending simple command with long output
        out = await iosxr.send_command("show run")
        print(out)
        # Testing sending configuration set
        commands = ["interface loopback 0", "description TEST LOOPBACK"]
        out = await iosxr.send_config_set(commands, with_commit=True)
        print(out)
        # Testing failed commit
        commands = ["interface GigabitEthernet 0/0/0/0", "service-policy input 1"]
        out = await iosxr.send_config_set(commands, with_commit=False)
        print(out)
        try:
            commands = ["interface GigabitEthernet 0/0/0/0", "service-policy input 2"]
            await iosxr.send_config_set(commands)
        except netdev.CommitError:
            print("Commit Error")


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'cisco_ios_xr']
    await asyncio.gather(*tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

IOS SG3XX example

#!/usr/bin/env python3

# Import Python library
import asyncio, netdev

# Coroutine used for the tasks
async def task(param):

    # Create an object for the devices and open SSH connections
    async with netdev.create(**param) as ios:

        # Testing sending simple command
        
        # Command to send
        cmd = "show clock"

        # Sending command
        output = await ios.send_command(cmd)

        # Display the output
        print(output)

# Main coroutine
async def main():

    # Parameters of the network device
    my_device = {   'username' : 'LOGIN',
                    'password' : 'PASSWORD',
                    'host': 'IP_ADDRESS',
                    'device_type': 'cisco_sg3xx',
    }

    # List of devices
    devices = [my_device]
    
    # List of tasks
    my_tasks = [task(dev) for dev in devices]
    
    # Starting the coroutine of the tasks
    await asyncio.wait(my_tasks)


# Main function call
if __name__ == '__main__':

    # Run the main coroutine
    asyncio.run(main())

    '''
    Result:
    ********************************************************************************
    .14:07:35 J  Aug 28 2020
    Time source is sntp
    Time from Browser is disabled
    ********************************************************************************

    '''

Fujitsu example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)

async def task(param):
    async with netdev.create(**param) as fuj:
        # Testing sending configuration set
        out = await fuj.send_config_set(['vlan database', 'exit'])
        print(out)
        # Testing sending simple command
        out = await fuj.send_command('show ver')
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'fujitsu_switch']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Comware example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as hp:
        # Testing sending simple command
        out = await hp.send_command('display ver')
        print(out)
        # Testing sending configuration set
        commands = ["Vlan 1", "quit"]
        out = await hp.send_config_set(commands)
        print(out)
        # Testing sending simple command with long output
        out = await hp.send_command('display cur')
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'hp_comware']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Comware Limited example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as hp:
        # Testing sending simple command
        out = await hp.send_command('display ver')
        print(out)
        # Testing sending configuration set
        commands = ["Vlan 1", "quit"]
        out = await hp.send_config_set(commands)
        print(out)
        # Testing sending simple command with long output
        out = await hp.send_command('display cur')
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'hp_comware_limited']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Mikrotik RouterOS example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as routeros:
        # Testing sending simple command
        commands = ['/ip address', 'print', '/']
        for cmd in commands:
            print(await routeros.send_command(cmd))

        # Testing sending configuration set
        out = await routeros.send_config_set(commands)
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'mikrotik_routeros']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Arista EOS example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as arista:
        # Testing sending simple command
        out = await arista.send_command('show run', strip_command=True)
        print(out)
        # Testing sending configuration set
        commands = ["vlan 1", "exit"]
        out = await arista.send_config_set(commands)
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'arista_eos']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Aruba AOS 6.X example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as ios:
        # Testing sending simple command
        out = await ios.send_command("show ver")
        print(out)
        # Testing sending configuration set
        commands = ["interface loopback", "exit"]
        out = await ios.send_config_set(commands)
        print(out)
        # Testing sending simple command with long output
        out = await ios.send_command("show run")
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'aruba_aos_6']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Juniper JunOS example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as junos:
        # Testing sending simple command
        out = await junos.send_command("show version")
        print(out)
        # Testing sending configuration set
        commands = ["edit system", "edit login"]
        out = await junos.send_config_set(commands, with_commit=True)
        print(out)
        # Testing sending simple command with long output
        out = await junos.send_command("show config")
        print(out)
        # Testing interactive dialog
        commands = ["set system login message 123", "delete system login message 123"]
        out = await junos.send_config_set(commands, with_commit=False, exit_config_mode=False)
        out += await junos.send_command("exit", pattern=r'Exit with uncommitted changes\?', strip_command=False)
        out += await junos.send_command("no", strip_command=False)
        out += await junos.send_command("rollback 0", strip_command=False)
        out += await junos.send_command("exit configuration-mode", strip_command=False)
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'juniper_junos']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Terminal example

import asyncio
import logging

import yaml

import netdev

config_path = 'config.yaml'

logging.basicConfig(level=logging.INFO)
netdev.logger.setLevel(logging.DEBUG)


async def task(param):
    async with netdev.create(**param) as term:
        # Testing sending simple command
        out = await term.send_command('ls -al', strip_command=True)
        print(out)
        out = await term.send_command('pwd', strip_command=True)
        print(out)
        out = await term.send_command('echo test', strip_command=True)
        print(out)


async def run():
    config = yaml.safe_load(open(config_path, 'r'))
    devices = yaml.safe_load(open(config['device_list'], 'r'))
    tasks = [task(dev) for dev in devices if dev['device_type'] == 'terminal']
    await asyncio.wait(tasks)


loop = asyncio.get_event_loop()
loop.run_until_complete(run())

Alcatel AOS example

#!/usr/bin/env python3

# Import Python library
import asyncio, netdev

# Coroutine used for the tasks
async def task(param):

    # Create an object for the devices and open SSH connections
    async with netdev.create(**param) as device:

        # Testing sending simple command
        
        # Command to send
        cmd = "show system"

        # Sending command
        output = await device.send_command(cmd)

        # Display the output
        print(output)

        # Display separator
        print("*" * 80)

        # Testing sending configuration set

        # Commands to send
        commands = ["vlan 3000", "no vlan 3001"]

        # Sending command
        output = await device.send_config_set(commands)

        # Display the output
        print(output)


# Main coroutine
async def main():

    # Parameters of the network device
    my_device = {   'username' : 'LOGIN',
                    'password' : 'PASSWORD',
                    'host': 'IP_ADDRESS',
                    'device_type': 'alcatel_aos',
    }

    # List of devices
    devices = [my_device]
    
    # List of tasks
    my_tasks = [task(dev) for dev in devices]
    
    # Starting the coroutine of the tasks
    await asyncio.wait(my_tasks)


# Main function call
if __name__ == '__main__':

    # Run the main coroutine
    asyncio.run(main())

    '''
    Result:
    ********************************************************************************
    System:
    Description:  Alcatel-Lucent Enterprise OS6860E-48 8.4.1.141.R03 GA, December 07, 2017.,
    Object ID:    1.3.6.1.4.1.6486.801.1.1.2.1.11.1.7,
    Up Time:      5 days 5 hours 3 minutes and 56 seconds,
    Contact:      Alcatel-Lucent, http://enterprise.alcatel-lucent.com,
    Name:         switch01,
    Location:     Somewhere nearby,
    Services:     78,
    Date & Time:  SAT AUG 29 2020 18:48:53 (CEST)
    Flash Space:
        Primary CMM:
        Available (bytes):  933896192,
        Comments         :  None

    ********************************************************************************
    vlan 3000

    switch01 ==> no vlan 3001

    switch01 ==>

    ********************************************************************************

    '''