[Day #51 PyATS Series] Custom Parser for Unsupported Cisco/Arista/PaloAlto/Fortigate Commands using pyATS [Python for Network Engineer]

• Trainer Sagar Dhawan
• Posted on September 6, 2025
• No Comments

[Day #51 PyATS Series] Custom Parser for Unsupported Cisco/Arista/PaloAlto/Fortigate Commands using pyATS [Python for Network Engineer]

---

Introduction — key points

When automating network validation with pyATS and Genie you often rely on built-in parsers (device.parse('show ...')). But real networks contain vendor commands, feature flags, or platforms that Genie does not parse yet — or you have a bespoke command that prints useful telemetry. Building a custom parser gives you reliable structured data from those raw CLI strings so your automation can assert, alert, and visualize.

In this Article you will learn to:

• Write MetaParser-based custom parsers (Genie style) for Cisco IOS-XE, Arista EOS, Palo Alto, and FortiGate outputs.
• Use regex, line-state machines, and table parsing strategies to extract fields.
• Integrate parsers into a pyATS job to run against your testbed.yml.
• Unit-test parsers against sample CLI output files.
• Validate results in CLI and GUI (Kibana/Grafana or simple Flask), store JSON artifacts and detect drift.
• Deploy, maintain and version parsers safely.

This is a full masterclass Article— all scripts are copy-paste runnable with the small edits noted. Let’s get into the lab.

---

Topology Overview

We’ll use a minimal multi-vendor lab for examples:

Devices must be reachable from the Automation Host via SSH. Parsers will consume CLI output of commands not parsed by Genie (or where you prefer a different structure).

---

Topology & Communications

What we will collect & parse (examples):

• Cisco IOS-XE: show platform hardware qfp active feature throughput (example unsupported command)
• Arista EOS: show hardware lldp neighbors detail (assume non-standard vendor output)
• Palo Alto: show running resource-monitor session (pseudo command)
• FortiGate: get system performance top (sample output not in Genie)

How we communicate:

• pyATS uses SSH via the Genie device object to run the raw command:
  raw = device.execute('show ...')
• We then feed raw into our custom parser class: parsed = ParserClass().cli(output=raw)
• Results are normalized JSON that downstream automation, alerts, or dashboards can consume.

Validation flows:

1. CLI: show the raw CLI, show parsed JSON, run assertions (e.g., peer_count == expected).
2. GUI: push parsed JSON to Elasticsearch and create Kibana visualizations (time-series of an extracted metric).
3. Unit tests: run parser against stored sample outputs to confirm behavior before running on live devices.

---

Workflow Script

Below is an end-to-end repo layout and the main runner script. I’ll include parser classes afterwards.

Repo layout (recommended)

pyats-custom-parsers/
├─ parsers/
│  ├─ __init__.py
│  ├─ cisco_custom.py
│  ├─ arista_custom.py
│  ├─ paloalto_custom.py
│  └─ fortigate_custom.py
├─ samples/
│  ├─ cisco_cmd_output.txt
│  ├─ arista_cmd_output.txt
│  └─ ...
├─ run_parsers.py
├─ testbed.yml
└─ tests/
   ├─ test_cisco_parser.py
   └─ ...

Main pyATS runner: run_parsers.py

#!/usr/bin/env python3
"""
run_parsers.py
Load testbed; for each device run the vendor-specific unsupported command,
parse using custom parser module, save JSON result for GUI/alerts.
"""

import json
from pathlib import Path
from genie.testbed import load
from parsers.cisco_custom import CiscoCustomParser
from parsers.arista_custom import AristaCustomParser
from parsers.paloalto_custom import PaloAltoCustomParser
from parsers.fortigate_custom import FortiGateCustomParser

OUT = Path('results')
OUT.mkdir(exist_ok=True)

PARSER_MAP = {
    'iosxe': CiscoCustomParser,
    'eos': AristaCustomParser,
    'panos': PaloAltoCustomParser,
    'fortios': FortiGateCustomParser
}

def get_parser_for_device(device):
    os_name = device.os.lower()
    for key, parser_cls in PARSER_MAP.items():
        if key in os_name:
            return parser_cls
    return None

def main():
    tb = load('testbed.yml')
    for name, device in tb.devices.items():
        print(f"[+] Processing {name} ({device.os})")
        parser_cls = get_parser_for_device(device)
        if not parser_cls:
            print(f"  - No parser for OS {device.os}, skipping.")
            continue
        # the raw command string should be defined in parser class for consistency
        raw_cmd = parser_cls.cli_command
        try:
            device.connect(log_stdout=False)
            raw_output = device.execute(raw_cmd)
            device.disconnect()
        except Exception as e:
            print(f"  - Failed to execute command on {name}: {e}")
            continue

        parser = parser_cls()
        # Attach device if parser needs device metadata
        parser.device = device
        parsed = parser.cli(output=raw_output)
        # store parsed JSON
        out_file = OUT / f"{name}_{parser_cls.__name__}.json"
        with open(out_file, 'w') as f:
            json.dump(parsed, f, indent=2)
        print(f"  - Parsed output saved to {out_file}")

if __name__ == "__main__":
    main()

This runner:

• Loads the testbed.yml
• Selects appropriate parser class by device.os
• Executes the cli_command defined inside each parser
• Instantiates the parser and calls cli(output=raw) to get structured data
• Writes JSON for downstream systems

---

Explanation by Line (runner)

• PARSER_MAP: map short OS identifiers to parser classes; add other mappings for additional vendors.
• get_parser_for_device: simple lookup — adjust for your device.os strings.
• parser_cls.cli_command: each parser class exposes cli_command so runner knows what raw command to execute (keeps runner generic).
• parser.cli(output=raw_output): our parsers implement cli() method following Genie metaparser pattern; they return a Python dict.
• Save results under results/ for GUI ingestion or manual inspection.

---

parsers/* Example Implementations

Below are complete parser implementations for four vendors. Put each in its own module under parsers/.

Note: All parser classes subclass genie.metaparser.MetaParser. This gives you schema discipline and consistent interface. We also show how to implement cli() and unit test behavior.

parsers/cisco_custom.py

# parsers/cisco_custom.py
import re
from genie.metaparser import MetaParser

class CiscoCustomParser(MetaParser):
    """
    Parser for a hypothetical unsupported Cisco command:
    show platform hardware qfp active feature throughput
    Expected raw lines (example):
      Feature    Peak(b/s)   Avg(b/s)   DropPct  Interface
      NAT        1200000     300000     0.2      TenGigE0/0/0
    """
    cli_command = 'show platform hardware qfp active feature throughput'

    schema = {
        'feature': {
            str: {
                'peak_bps': int,
                'avg_bps': int,
                'drop_pct': float,
                'interface': str
            }
        }
    }

    def cli(self, output=None):
        import math
        if output is None:
            output = self.device.execute(self.cli_command)
        ret = {'feature': {}}
        # Skip header lines until we hit a data line
        for line in output.splitlines():
            line = line.strip()
            if not line:
                continue
            # regex to parse the columns (loose spacing)
            m = re.match(r'(?P<feature>\S+)\s+(?P<peak>\d+)\s+(?P<avg>\d+)\s+(?P<drop>[\d\.]+)\s+(?P<intf>\S+)', line)
            if m:
                g = m.groupdict()
                ret['feature'][g['feature']] = {
                    'peak_bps': int(g['peak']),
                    'avg_bps': int(g['avg']),
                    'drop_pct': float(g['drop']),
                    'interface': g['intf']
                }
        return ret

parsers/arista_custom.py

# parsers/arista_custom.py
import re
from genie.metaparser import MetaParser

class AristaCustomParser(MetaParser):
    """
    Example parser for 'show hardware lldp neighbors detail' (nonstandard format)
    """
    cli_command = 'show hardware lldp neighbors detail'

    schema = {
        'neighbors': {
            str: {  # local interface
                'remote_system': str,
                'remote_port': str,
                'ttl': int
            }
        }
    }

    def cli(self, output=None):
        if output is None:
            output = self.device.execute(self.cli_command)
        neighbors = {}
        current_intf = None
        for line in output.splitlines():
            line = line.strip()
            # e.g. "Local Intf: Ethernet1" or "Local Intf: Et1"
            m = re.match(r'Local Intf:\s*(?P<intf>\S+)', line)
            if m:
                current_intf = m.group('intf')
                continue
            if current_intf:
                m2 = re.match(r'Remote System:\s*(?P<sys>.+)', line)
                if m2:
                    neighbors.setdefault(current_intf, {})['remote_system'] = m2.group('sys')
                    continue
                m3 = re.match(r'Remote Port:\s*(?P<port>\S+)', line)
                if m3:
                    neighbors.setdefault(current_intf, {})['remote_port'] = m3.group('port')
                    continue
                m4 = re.match(r'TTL:\s*(?P<ttl>\d+)', line)
                if m4:
                    neighbors.setdefault(current_intf, {})['ttl'] = int(m4.group('ttl'))
                    continue
        return {'neighbors': neighbors}

parsers/paloalto_custom.py

# parsers/paloalto_custom.py
import re
from genie.metaparser import MetaParser

class PaloAltoCustomParser(MetaParser):
    """
    Parser for a pseudo PAN-OS command output showing session summary.
    """
    cli_command = 'show running resource-monitor session'

    schema = {
        'sessions': {
            'total': int,
            'tcp': int,
            'udp': int,
            'icmp': int
        }
    }

    def cli(self, output=None):
        if output is None:
            output = self.device.execute(self.cli_command)
        data = {'sessions': {}}
        for line in output.splitlines():
            line = line.strip()
            m_total = re.match(r'Total Sessions:\s*(\d+)', line)
            if m_total:
                data['sessions']['total'] = int(m_total.group(1))
            m_tcp = re.match(r'TCP Sessions:\s*(\d+)', line)
            if m_tcp:
                data['sessions']['tcp'] = int(m_tcp.group(1))
            m_udp = re.match(r'UDP Sessions:\s*(\d+)', line)
            if m_udp:
                data['sessions']['udp'] = int(m_udp.group(1))
            m_icmp = re.match(r'ICMP Sessions:\s*(\d+)', line)
            if m_icmp:
                data['sessions']['icmp'] = int(m_icmp.group(1))
        return data

parsers/fortigate_custom.py

# parsers/fortigate_custom.py
import re
from genie.metaparser import MetaParser

class FortiGateCustomParser(MetaParser):
    """
    Parse a FortiGate 'get system performance top' style summary into structured fields.
    """
    cli_command = 'get system performance top'

    schema = {
        'cpu': {
            'usage_percent': float,
            'spikes_1m': int
        },
        'mem': {
            'total_mb': int,
            'used_mb': int
        }
    }

    def cli(self, output=None):
        if output is None:
            output = self.device.execute(self.cli_command)

        out = {'cpu': {}, 'mem': {}}
        for line in output.splitlines():
            line = line.strip()
            m_cpu = re.match(r'CPU usage:\s*([\d\.]+)%\s+spikes:\s*(\d+)', line, re.IGNORECASE)
            if m_cpu:
                out['cpu']['usage_percent'] = float(m_cpu.group(1))
                out['cpu']['spikes_1m'] = int(m_cpu.group(2))
                continue
            m_mem = re.match(r'Memory:\s*(\d+)MB total,\s*(\d+)MB used', line, re.IGNORECASE)
            if m_mem:
                out['mem']['total_mb'] = int(m_mem.group(1))
                out['mem']['used_mb'] = int(m_mem.group(2))
                continue
        return out

---

testbed.yml Example

Place the testbed next to run_parsers.py. For safety, in labs use test accounts.

testbed:
  name: custom_parser_lab
  credentials:
    default:
      username: netdev
      password: netdev123
  devices:
    R1:
      os: iosxe
      type: router
      connections:
        cli:
          protocol: ssh
          ip: 10.0.100.11
    A1:
      os: eos
      type: switch
      connections:
        cli:
          protocol: ssh
          ip: 10.0.100.21
    PA1:
      os: panos
      type: firewall
      connections:
        cli:
          protocol: ssh
          ip: 10.0.100.31
    FG1:
      os: fortios
      type: firewall
      connections:
        cli:
          protocol: ssh
          ip: 10.0.100.41

---

Post-validation CLI (Real expected output)

Below are sample raw outputs (what the device returns) and expected parsed JSON produced by the parsers.

A — Cisco raw output (show platform hardware qfp active feature throughput)

Feature    Peak(b/s)   Avg(b/s)   DropPct  Interface
NAT        1200000     300000     0.2      TenGigE0/0/0
Firewall   500000      120000     1.5      TenGigE0/0/1

Parsed JSON (CiscoCustomParser)

{
  "feature": {
    "NAT": {
      "peak_bps": 1200000,
      "avg_bps": 300000,
      "drop_pct": 0.2,
      "interface": "TenGigE0/0/0"
    },
    "Firewall": {
      "peak_bps": 500000,
      "avg_bps": 120000,
      "drop_pct": 1.5,
      "interface": "TenGigE0/0/1"
    }
  }
}

B — Arista raw output

Local Intf: Ethernet1
  Remote System: core-switch.example.com
  Remote Port: Eth2
  TTL: 120

Local Intf: Ethernet2
  Remote System: server.example.net
  Remote Port: Eth1
  TTL: 120

Parsed JSON (AristaCustomParser)

{
  "neighbors": {
    "Ethernet1": {"remote_system": "core-switch.example.com", "remote_port": "Eth2", "ttl": 120},
    "Ethernet2": {"remote_system": "server.example.net", "remote_port": "Eth1", "ttl": 120}
  }
}

C — Palo Alto raw output

Total Sessions: 1200
TCP Sessions: 800
UDP Sessions: 350
ICMP Sessions: 50

Parsed JSON (PaloAltoCustomParser)

{
  "sessions": {"total": 1200, "tcp": 800, "udp": 350, "icmp": 50}
}

D — FortiGate raw output

CPU usage: 12.5% spikes: 0
Memory: 4096MB total, 2048MB used

Parsed JSON (FortiGateCustomParser)

{
  "cpu": {"usage_percent": 12.5, "spikes_1m": 0},
  "mem": {"total_mb": 4096, "used_mb": 2048}
}

---

Unit Testing Parsers (best practice)

Always test parsers with representative sample outputs before running them against live devices.

Example pytest test: tests/test_cisco_parser.py

from parsers.cisco_custom import CiscoCustomParser

def test_cisco_parser_with_sample():
    sample = open('samples/cisco_cmd_output.txt').read()
    p = CiscoCustomParser()
    parsed = p.cli(output=sample)
    assert 'feature' in parsed
    assert parsed['feature']['NAT']['peak_bps'] == 1200000

Run: pytest -q tests/

Unit tests help prevent regressions when CLI output formats change after upgrades.

---

Explanation by Line (parser internals & patterns)

Key parsing strategies used:

1. Columnar table parsing with regex:
   When output is a clean table, use a regex matching columns (\s+ separators) and convert to typed fields (int, float). The Cisco parser demonstrates this.
2. Stateful block parsing:
   For outputs that contain blocks per interface or neighbor, iterate lines and switch state when you see a block header (Local Intf:). The Arista parser uses that.
3. Key: value extraction:
   For simple key/value outputs (Palo Alto), scan lines for patterns like ^Key:\s+Value and map them.
4. Type conversions & validation:
   Always convert captured strings to int / float and handle missing values gracefully.
5. Device context:
   Attach parser.device = device if you need device metadata (hostname, platform) in parser logic (e.g., to enrich output with site tag).
6. Schema enforcement:
   Use schema attribute so your parser documents expected fields — this is helpful for later validation and unit tests. The parser returns a dict matching the schema.

Defensive parsing tips:

• Be tolerant: CLI can vary by software version. Try multiple regex patterns (fallbacks).
• Sanitize: remove ANSI color codes, leading/trailing whitespace, and control characters.
• Logging: add debug logs in parser code while developing — use logging module.

---

GUI Integration & Validation Flow

Once run_parsers.py saves JSON files to results/, you can:

• Push them to Elasticsearch index custom-parsers-* (one doc per run/device) for historical trend analysis.
• Create Kibana dashboards: e.g., time-series of cpu.usage_percent from FortiGate, or sessions.total from Palo Alto.
• Alert: create Elasticsearch Watcher or Kibana Alert to notify when sessions.total > threshold or cpu.usage_percent > 80.

Minimal Flask dashboard (quick example) — show last-parsed JSON files:

# app.py (Flask)
from flask import Flask, render_template, jsonify
import glob, json

app = Flask(__name__)

@app.route('/')
def index():
    files = sorted(glob.glob('results/*.json'), reverse=True)[:50]
    docs = []
    for f in files:
        docs.append(json.load(open(f)))
    return jsonify(docs)

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=8080)

This simple GUI is handy for workshops; use ES for production.

---

Maintenance & Deployment

Where to place custom parsers in production:

• Keep parsers/ as a Python package and include it in your automation virtualenv or deploy it as a wheel (pip install .).
• Optionally contribute general parsers to the Genie community if they’re broadly useful.

Version control:

• Store parsers/ and samples/ in Git.
• Use CI to run parser unit tests on PRs (pytest).
• When device OS upgrades change CLI, update parser regex and add new samples.

Performance:

• Parsers run very fast; main limit is SSH execution time.
• For many devices, run parser jobs concurrently (ThreadPoolExecutor or pyATS parallel features).

---

FAQs

1. When should I build a custom parser vs using TextFSM / NTC templates?

Answer: Use Genie custom parsers when you want tight schema enforcement and integration with pyATS/Genie. TextFSM (ntc-templates) is great for quick table parsing but returns lists of lists and requires template management. If you already use pyATS/Genie, a MetaParser class gives better structured output and unit testing support.

---

2. How do I handle multiline entries (e.g., long BGP route attributes) in parsers?

Answer: Use a state machine: detect block headers, collect indented lines into a buffer until next header, then parse the whole block. Regex with re.DOTALL can also help but state machines are more readable and maintainable.

---

3. How do I test a parser against multiple platform versions?

Answer: Store sample outputs from each OS/version in samples/ and create pytest cases for each. Run unit tests in CI on every PR. Add regression samples when you discover format changes.

---

4. How to handle vendor CLI differences in the same parser?

Answer: Either write vendor-specific parser classes (preferred: smaller, easier) or create a shared base parser that branches on self.device.os or platform to apply different regex patterns.

---

5. Is it safe to call device.parse() with my custom parser?

Answer: Genie will only use its built-in parser discovery for installed parsers. For local custom parsers, either register them in the Genie namespace or instantiate and call .cli(output=raw) directly as shown — this is the safest and most explicit method.

---

6. How do I ensure parsers won’t break production?

Answer: Unit test thoroughly. Run parsers against sampled outputs after device upgrades. Use try/except in runner scripts so a parser exception for one device doesn’t abort the whole job.

---

7. Can custom parsers be used in pyATS tests and testcases (pass/fail assertions)?

Answer: Yes. Parse into a dict and use pyats.aetest.Testcase to assert values. Example: self.assertTrue(parsed['sessions']['total'] < threshold, 'High session count').

---

8. How do you debug a failing parser on live data?

Answer: Capture raw output to samples/ (raw_output.dump) and run parser locally against it. Add debug print() or logging.debug() to trace regex groups. Maintain a sample corpus to reproduce edge cases.

---

YouTube Link

Watch the Complete Python for Network Engineer: Custom parser for unsupported Cisco/Arista/Paloalto/Fortigate commands Using pyATS for Cisco [Python for Network Engineer] Lab Demo & Explanation on our channel:

---

Join Our Training

If you want step-by-step, instructor-led training that walks you through building production-grade parsers, CI-tested automation, GUI dashboards, and real-world projects, join Trainer Sagar Dhawan’s 3-month instructor-led program covering Python, Ansible, APIs, and Cisco DevNet for Network Engineers.

This course teaches exactly how to move from CLI hacks to robust automation — specifically targeted at the Python for Network Engineer path. You’ll get hands-on labs, code reviews, and career guidance.

Enroll / learn more:
https://course.networkjourney.com/python-ansible-api-cisco-devnet-for-network-engineers/

Enroll Now & Future‑Proof Your Career
Email: info@networkjourney.com
WhatsApp / Call: +91 97395 21088

---