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())

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())