# py312
# requirements:
# pyeapi==1.0.4
import ssl
import time
from typing import Literal, TypedDict
from urllib.parse import urlsplit
import pyeapi
class arista_eos(TypedDict):
host: str
username: str
password: str
port: int
verify_tls: bool
def main(
auth: arista_eos,
lines: list[str],
mode: Literal["diff", "commit", "commit_timer"] = "diff",
session_name: str = "",
timer_minutes: int = 10,
):
"""
Push Config
Apply configuration lines in an EOS config session and return the diff. mode diff only shows the diff and aborts (dry run); commit applies it; commit_timer applies it and rolls it back by itself unless Confirm Commit runs within timer_minutes. Nothing is saved to startup-config: run Save Config (or Confirm Commit) for that.
"""
session = session_name or f"windmill-{int(time.time() * 1000)}"
# A session survives across eAPI calls by name: stage the lines, read the diff, then commit or abort it.
eos = node(auth)
eos.run_commands([f"configure session {session}", *lines, "end"])
[diff] = eos.run_commands([f"show session-config named {session} diffs"], "text")
if mode == "commit":
eos.run_commands([f"configure session {session} commit"])
elif mode == "commit_timer":
m = timer_minutes or 10
eos.run_commands(
[f"configure session {session} commit timer {m // 60:02d}:{m % 60:02d}:00"],
)
else:
eos.run_commands([f"configure session {session} abort"])
return {
"session": session,
"mode": mode or "diff",
"diff": diff["output"],
"committed": mode in ("commit", "commit_timer"),
}
# host is an address or a URL (https://10.0.0.11:8443). pyeapi skips certificate checks unless it is given a context,
# so one is always passed: verifying, or not when the resource unticks verify_tls (EOS ships a self-signed cert).
def node(auth: arista_eos, timeout: float = 60):
url = urlsplit(auth["host"] if "://" in auth["host"] else f"https://{auth['host']}")
context = ssl.create_default_context()
if auth.get("verify_tls") is False:
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
return pyeapi.connect(
transport=url.scheme,
host=url.hostname,
port=url.port or auth.get("port") or None,
username=auth["username"],
password=auth["password"],
context=context,
timeout=timeout,
return_node=True,
)
Submitted by hugo989 3 hours ago