Python в DevOps: автоматизация сетевых устройств

Расширенные примеры автоматизации сети с Python

Пример 1: Резервное копирование конфигураций коммутаторов Cisco с помощью Netmiko и Jinja2

Создадим скрипт, который подключается к списку устройств, забирает running-config и сохраняет в файл с именем, содержащим дату и имя хоста. Для формирования имени используется шаблон Jinja2.

from netmiko import ConnectHandler
from jinja2 import Template
from datetime import datetime
import os

devices = [
    {'host': '192.168.1.10', 'device_type': 'cisco_ios', 'username': 'admin', 'password': 'secret'},
    {'host': '192.168.1.20', 'device_type': 'cisco_ios', 'username': 'admin', 'password': 'secret'}
]

backup_dir = 'backups'
os.makedirs(backup_dir, exist_ok=True)

filename_template = Template('backup_{{ hostname }}_{{ date }}.txt')
date_str = datetime.now().strftime('%Y%m%d_%H%M%S')

for device in devices:
    try:
        with ConnectHandler(**device) as conn:
            hostname = conn.send_command('show running-config | include hostname').split()[1]
            config = conn.send_command('show running-config')
            filename = filename_template.render(hostname=hostname, date=date_str)
            filepath = os.path.join(backup_dir, filename)
            with open(filepath, 'w') as f:
                f.write(config)
            print(f'Конфигурация {hostname} сохранена в {filepath}')
    except Exception as e:
        print(f'Ошибка при работе с {device["host"]}: {e}')

Конфигурация Switch-A сохранена в backups/backup_Switch-A_20250315_120500.txt
Конфигурация Router-B сохранена в backups/backup_Router-B_20250315_120500.txt

В скрипте используется Jinja2 для динамического формирования имени файла. Обработка исключений позволяет продолжить резервное копирование даже при недоступности одного из устройств.

Пример 2: Параллельный опрос устройств с помощью асинхронного Scrapli

Задача: получить uptime со всех коммутаторов в сети одновременно. Используется asyncio и Scrapli AsyncIOSXEDriver.

import asyncio
from scrapli.driver.core import AsyncIOSXEDriver

devices_params = [
    {'host': '192.168.1.10', 'auth_username': 'admin', 'auth_password': 'secret', 'transport': 'ssh2'},
    {'host': '192.168.1.11', 'auth_username': 'admin', 'auth_password': 'secret', 'transport': 'ssh2'},
]

async def get_uptime(params):
    async with AsyncIOSXEDriver(**params) as conn:
        response = await conn.send_command('show version | include uptime')
        return params['host'], response.result

async def main():
    tasks = [get_uptime(p) for p in devices_params]
    for future in asyncio.as_completed(tasks):
        host, result = await future
        print(f'{host}: {result.strip()}')

asyncio.run(main())

192.168.1.10: Switch uptime is 2 weeks, 3 days, 5 hours, 22 minutes
192.168.1.11: Router uptime is 1 week, 6 days, 12 hours, 10 minutes

Асинхронный подход сокращает общее время ожидания – устройства опрашиваются параллельно, а не последовательно.

Пример 3: Сравнение конфигураций с помощью NAPALM

Напишем скрипт, который загружает текущую конфигурацию устройства, затем предлагает изменения (например, добавить новый VLAN) и показывает разницу до коммита.

from napalm import get_network_driver
from napalm.base.exceptions import MergeError

driver = get_network_driver('ios')
with driver('192.168.1.10', 'admin', 'secret') as device:
    config_before = device.get_config()['running']
    
    # Подготовка изменений (например, новый VLAN)
    candidate = (
        'vlan 100\n'
        ' name TestVLAN\n'
    )
    
    device.load_merge_candidate(config=candidate)
    diff = device.compare_config()
    if diff:
        print('Изменения, которые будут применены:')
        print(diff)
        # При необходимости:
        # device.commit()
        # device.rollback()
    else:
        print('Изменения не требуются')

Изменения, которые будут применены:
[vlan]
+ vlan 100
+  name TestVLAN

Метод compare_config возвращает разницу в формате unified diff. Это удобно для аудита перед применением изменений.

Пример 4: Взаимодействие с Meraki Dashboard API через requests с обработкой пагинации

Получим список всех устройств (здесь – коммутаторов) в организации Meraki с помощью постраничного запроса.

import requests

api_key = 'YOUR_API_KEY'
org_id = 'YOUR_ORG_ID'
base_url = f'https://api.meraki.com/api/v1/organizations/{org_id}/devices'

headers = {
    'X-Cisco-Meraki-API-Key': api_key,
    'Accept': 'application/json'
}

all_devices = []
url = base_url

while url:
    response = requests.get(url, headers=headers)
    if response.status_code != 200:
        print(f'Ошибка: {response.status_code}')
        break
    data = response.json()
    all_devices.extend(data)
    
    # Обработка пагинации (Link header)
    link_header = response.headers.get('Link', '')
    next_url = None
    if 'rel="next"' in link_header:
        for part in link_header.split(','):
            if 'rel="next"' in part:
                next_url = part[part.index('<')+1:part.index('>')]
                break
    url = next_url

print(f'Найдено устройств: {len(all_devices)}')
for dev in all_devices[:3]:
    print(f'{dev["name"]} - {dev["model"]} - {dev["serial"]}')

Найдено устройств: 12
Switch-A - MS250-48 - Q2CD-XXXX
Switch-B - MS225-24 - Q2CE-YYYY
Router-C - MX64 - Q2CF-ZZZZ

Скрипт обрабатывает пагинацию через заголовок Link. Это важно для организаций с большим количеством устройств, чтобы получить полный список.