A Reliable Network Automation Workflow

import csv
import json
import logging
import time

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

REQUIRED_NTP_SERVER = "10.0.0.5"
DRY_RUN = True  # default: never changes a real device unless explicitly set to False


def is_valid_device(device):
  """A device record needs at least a hostname and an IP to be usable."""
  return bool(device.get("hostname")) and bool(device.get("ip"))


class MockConnection:
  """Stands in for a real Netmiko connection -- see 'apply_ntp_config' below."""

  def __init__(self, already_has_ntp):
      self.configured = already_has_ntp

  def send_command(self, cmd):
      return f"ntp server {REQUIRED_NTP_SERVER}" if self.configured else ""

  def send_config_set(self, lines):
      self.configured = True
      return "ok"


# Option B, no device needed: a deterministic scenario per hostname.
# Option A swaps get_connection() for the real Netmiko version shown below.
_MOCK_SCENARIOS = {
  "SW1": {"fails_before_success": 0, "auth_fails": False, "already_has_ntp": True},
  "SW2": {"fails_before_success": 0, "auth_fails": False, "already_has_ntp": False},
  "SW3": {"fails_before_success": 2, "auth_fails": False, "already_has_ntp": False},
  "SW4": {"fails_before_success": 99, "auth_fails": False, "already_has_ntp": False},
  "SW5": {"fails_before_success": 0, "auth_fails": True, "already_has_ntp": False},
}
_attempt_counts = {}
_MOCK_CONNECTIONS = {h: MockConnection(s["already_has_ntp"]) for h, s in _MOCK_SCENARIOS.items()}


def get_connection(device):
  hostname = device["hostname"]
  scenario = _MOCK_SCENARIOS[hostname]
  if scenario["auth_fails"]:
      raise PermissionError(f"{hostname} rejected the credentials")
  _attempt_counts[hostname] = _attempt_counts.get(hostname, 0) + 1
  if _attempt_counts[hostname] <= scenario["fails_before_success"]:
      raise TimeoutError(f"{hostname} did not respond")
  return _MOCK_CONNECTIONS[hostname]


def connect_with_retry(device, max_attempts=3, delay=2):
  """Retries a timeout up to max_attempts times; never retries an authentication failure."""
  for attempt in range(1, max_attempts + 1):
      try:
          return get_connection(device), attempt
      except TimeoutError:
          if attempt == max_attempts:
              raise
          time.sleep(delay)
      except PermissionError:
          raise


def apply_ntp_config(connection, required_ntp_server, dry_run=True):
  """Pre-check, change only if needed, and independently post-check to verify."""
  pre_check = connection.send_command("show run | include ntp server")
  already_configured = required_ntp_server in pre_check

  if already_configured:
      return {"changed": False, "verified": True, "dry_run": dry_run}

  if dry_run:
      return {"changed": False, "verified": None, "dry_run": True}

  connection.send_config_set([f"ntp server {required_ntp_server}"])
  post_check = connection.send_command("show run | include ntp server")
  verified = required_ntp_server in post_check
  return {"changed": True, "verified": verified, "dry_run": False}


def run_device(device, dry_run=True):
  hostname = device["hostname"]
  try:
      connection, attempts = connect_with_retry(device)
  except TimeoutError:
      logging.warning(f"{hostname}: unreachable after 3 attempts")
      return {"hostname": hostname, "ip": device["ip"], "status": "unreachable", "attempts": 3, "retried": True}
  except PermissionError:
      logging.warning(f"{hostname}: authentication failed")
      return {"hostname": hostname, "ip": device["ip"], "status": "auth-failed", "attempts": 1, "retried": False}

  outcome = apply_ntp_config(connection, REQUIRED_NTP_SERVER, dry_run=dry_run)
  retried = attempts > 1

  if outcome["dry_run"] and not outcome["changed"] and outcome["verified"] is None:
      status = "dry-run-preview"
  elif outcome["changed"]:
      status = "changed" if outcome["verified"] else "change-not-verified"
  else:
      status = "already-compliant"

  logging.info(f"{hostname}: {status} (attempts={attempts})")
  return {"hostname": hostname, "ip": device["ip"], "status": status, "attempts": attempts, "retried": retried}


with open("devices.csv") as f:
  devices = list(csv.DictReader(f))

valid_devices = []
for device in devices:
  if is_valid_device(device):
      valid_devices.append(device)
  else:
      print(f"Skipping invalid inventory record: {device}")

results = [run_device(device, dry_run=DRY_RUN) for device in valid_devices]

with open("reliability_report.json", "w") as f:
  json.dump(results, f, indent=4)

by_status = {}
for r in results:
  by_status[r["status"]] = by_status.get(r["status"], 0) + 1

print("\n--- Summary ---")
print(f"Checked: {len(results)} devices")
for status, count in by_status.items():
  print(f"{status}: {count}")
print("Report written to reliability_report.json")

Save this as reliable_health_check.py alongside devices.csv and run 'python reliable_health_check.py' — genuinely complete and runnable as shown, using the built-in simulation (Option B). get_connection() is the one piece Option A replaces, shown next.

hostname,ip,platform
SW1,10.10.10.1,cisco_ios
SW2,10.10.10.2,cisco_ios
SW3,10.10.10.3,cisco_ios
SW4,10.10.10.4,cisco_ios
SW5,10.10.10.5,cisco_ios

Five devices this time, deliberately, so the report can show every outcome this module built: already correct, needs a change, needs a retry, unreachable, and rejected credentials.

from netmiko import ConnectHandler, NetmikoTimeoutException, NetmikoAuthenticationException

def get_connection(device):
  """Replace the mock get_connection() above with this to use a real device."""
  try:
      return ConnectHandler(
          device_type=device["platform"],
          host=device["ip"],
          username=USERNAME,   # supplied via input()/getpass(), never hardcoded
          password=PASSWORD,
          conn_timeout=10,
      )
  except NetmikoTimeoutException:
      raise TimeoutError(f"{device['hostname']} did not respond")
  except NetmikoAuthenticationException:
      raise PermissionError(f"{device['hostname']} rejected the credentials")

Same function name, same two exceptions raised (TimeoutError, PermissionError) — connect_with_retry() and everything after it doesn't need to know or care which version of get_connection() is running underneath it.

2026-08-25 22:23:41 INFO SW1: already-compliant (attempts=1)
2026-08-25 22:23:41 INFO SW2: dry-run-preview (attempts=1)
2026-08-25 22:23:41 INFO SW3: dry-run-preview (attempts=3)
2026-08-25 22:23:41 WARNING SW4: unreachable after 3 attempts
2026-08-25 22:23:41 WARNING SW5: authentication failed

--- Summary ---
Checked: 5 devices
already-compliant: 1
dry-run-preview: 2
unreachable: 1
auth-failed: 1
Report written to reliability_report.json

SW1 was already correct. SW2 would need a change — previewed, not applied. SW3 needed two retries before it even got that far, then would also need a change. SW4 exhausted its retries. SW5's credentials were rejected on the first try, no retry attempted.

[
  {
      "hostname": "SW1",
      "ip": "10.10.10.1",
      "status": "already-compliant",
      "attempts": 1,
      "retried": false
  },
  {
      "hostname": "SW2",
      "ip": "10.10.10.2",
      "status": "dry-run-preview",
      "attempts": 1,
      "retried": false
  },
  {
      "hostname": "SW3",
      "ip": "10.10.10.3",
      "status": "dry-run-preview",
      "attempts": 3,
      "retried": true
  },
  {
      "hostname": "SW4",
      "ip": "10.10.10.4",
      "status": "unreachable",
      "attempts": 3,
      "retried": true
  },
  {
      "hostname": "SW5",
      "ip": "10.10.10.5",
      "status": "auth-failed",
      "attempts": 1,
      "retried": false
  }
]

Every device has exactly one entry, with enough detail to answer 'what happened, and how many tries did it take?' — not just a bare status.