aos module
Management of AOS managed device and system-agents: - Managed Devices - System Agents - Device Profiles
Source code in aos/devices.py
class AosDevices(AosSubsystem):
"""
Management of AOS managed device and system-agents:
- Managed Devices
- System Agents
- Device Profiles
"""
def __init__(self, rest):
self.managed_devices = AosManagedDevices(rest)
self.system_agents = AosSystemAgents(rest)
self.device_profiles = AosDeviceProfiles(rest)
Management of AOS controlled devices
Source code in aos/devices.py
class AosManagedDevices(AosSubsystem):
"""
Management of AOS controlled devices
"""
def accept_running_config_as_golden(self, system_id: str):
self.rest.json_resp_post(
f"/api/systems/{system_id}/accept-running-config-as-golden"
)
def get_all(self) -> List[System]:
return list(self.iter_all())
def iter_all(self) -> Generator[System, None, None]:
systems = self.rest.json_resp_get("/api/systems")
for s in systems.get("items", []):
yield System.from_json(s)
def get_system_by_id(self, system_id: str) -> Optional[System]:
return System.from_json(self.rest.json_resp_get(f"/api/systems/{system_id}"))
def iter_anomalies(self, system_id: str) -> Generator[Anomaly, None, None]:
anomalies = self.rest.json_resp_get(f"api/systems/{system_id}/anomalies")
if anomalies is None:
return
for anomaly in anomalies["items"]:
yield Anomaly.from_json(anomaly)
def get_anomalies(self, system_id: str) -> List[Anomaly]:
return list(self.iter_anomalies(system_id))
def has_anomalies(self, system_id: str) -> bool:
return self.get_anomalies(system_id) != []
def has_anomalies_of_type(self, system_id: str, anomaly_type: str) -> bool:
for anomaly in self.iter_anomalies(system_id):
if anomaly.type == anomaly_type:
return True
return False
def find_system_with_ip(self, ip_addr: str) -> Optional[System]:
for system in self.iter_all():
if system.facts["mgmt_ipaddr"] == ip_addr:
return system
return NullSystem
def delete(self, agent_uuid: str) -> None:
self.rest.delete(f"/api/system-agents/{agent_uuid}")
def get_by_id(self, agent_uuid: str) -> Optional[SystemAgent]:
resp = self.rest.json_resp_get(f"/api/system-agents/{agent_uuid}")
if resp:
return SystemAgent.from_json(resp)
Management of system-agent for AOS controlled devices
Source code in aos/devices.py
class AosSystemAgents(AosSubsystem):
"""
Management of system-agent for AOS controlled devices
"""
def create_system_agent(self, data) -> bool:
return self.rest.json_resp_post("/api/system-agents", data=data)
def get_all(self) -> List[SystemAgent]:
return list(self.iter_all())
def iter_all(self) -> Generator[SystemAgent, None, None]:
system_agents = self.rest.json_resp_get("/api/system-agents")
for s in system_agents.get("items", []):
yield SystemAgent.from_json(s)
def get_agent_by_id(self, system_id: str) -> Optional[System]:
return System.from_json(
self.rest.json_resp_get(f"/api/system-agents/{system_id}")
)
def find_agent_with_ip(self, ip_addr: str) -> Optional[SystemAgent]:
for agent in self.iter_all():
if agent.management_ip == ip_addr:
return agent
return NullSystemAgent
def get_packages(self):
"""
Get a list of all device packages imported into AOS
"""
p_path = "/api/packages"
resp = self.rest.json_resp_get(p_path)
return [
DevicePackage(name=package["name"], version=package["version"])
for package in resp["items"]
]
def get_os_images(self):
"""
Get a list of all OS images imported into AOS
"""
p_path = "/api/device-os/images"
resp = self.rest.json_resp_get(p_path)
return [
DeviceOSImage(
description=image["description"],
checksum=image["checksum"],
image_name=image["image_name"],
platform=image["platform"],
image_url=image["image_url"],
type=image["type"],
id=image["id"],
)
for image in resp["items"]
]
def create(
self,
management_ip: str,
label: str,
username: str,
password: str,
platform: Optional[str] = None,
telemetry_only: bool = False,
is_offbox: bool = False,
) -> str:
"""
Creates system agent in AOS.
Parameters
----------
management_ip
(str) management IP address for system
label
(str) unique device identifier or name
username
(str) username for authentication on target device
password
(str) password for authentication on target device
platform
(str) (optional) system platform of target device can be one of:
[junos, nxos, eos]
Only applicable to offbox agents, ignored for onbox agents.
default: None
telemetry_only
(bool) (optional) AOS agent operation mode, Default sets agent
to full_control
default: False
is_offbox
(bool) (optional) setup offbox agent instead of directly on target device
default: False
Return
------
UUID of created system agent
"""
sys_agent = {
"agent_type": "offbox" if is_offbox else "onbox",
"job_on_create": "check",
"management_ip": management_ip,
"label": label,
"open_options": {},
"operation_mode": (
"telemetry_only" if telemetry_only else "full_control"
),
"password": password,
"username": username,
}
if is_offbox:
sys_agent["platform"] = platform
resp = self.rest.json_resp_post("/api/system-agents", data=sys_agent)
return resp["id"]
create(self, management_ip, label, username, password, platform=None, telemetry_only=False, is_offbox=False)
Creates system agent in AOS. Parameters
management_ip (str) management IP address for system label (str) unique device identifier or name username (str) username for authentication on target device password (str) password for authentication on target device platform (str) (optional) system platform of target device can be one of: [junos, nxos, eos] Only applicable to offbox agents, ignored for onbox agents. default: None telemetry_only (bool) (optional) AOS agent operation mode, Default sets agent to full_control default: False is_offbox (bool) (optional) setup offbox agent instead of directly on target device default: False Return
UUID of created system agent
Source code in aos/devices.py
def create(
self,
management_ip: str,
label: str,
username: str,
password: str,
platform: Optional[str] = None,
telemetry_only: bool = False,
is_offbox: bool = False,
) -> str:
"""
Creates system agent in AOS.
Parameters
----------
management_ip
(str) management IP address for system
label
(str) unique device identifier or name
username
(str) username for authentication on target device
password
(str) password for authentication on target device
platform
(str) (optional) system platform of target device can be one of:
[junos, nxos, eos]
Only applicable to offbox agents, ignored for onbox agents.
default: None
telemetry_only
(bool) (optional) AOS agent operation mode, Default sets agent
to full_control
default: False
is_offbox
(bool) (optional) setup offbox agent instead of directly on target device
default: False
Return
------
UUID of created system agent
"""
sys_agent = {
"agent_type": "offbox" if is_offbox else "onbox",
"job_on_create": "check",
"management_ip": management_ip,
"label": label,
"open_options": {},
"operation_mode": (
"telemetry_only" if telemetry_only else "full_control"
),
"password": password,
"username": username,
}
if is_offbox:
sys_agent["platform"] = platform
resp = self.rest.json_resp_post("/api/system-agents", data=sys_agent)
return resp["id"]
get_os_images(self)
Get a list of all OS images imported into AOS
Source code in aos/devices.py
def get_os_images(self):
"""
Get a list of all OS images imported into AOS
"""
p_path = "/api/device-os/images"
resp = self.rest.json_resp_get(p_path)
return [
DeviceOSImage(
description=image["description"],
checksum=image["checksum"],
image_name=image["image_name"],
platform=image["platform"],
image_url=image["image_url"],
type=image["type"],
id=image["id"],
)
for image in resp["items"]
]
get_packages(self)
Get a list of all device packages imported into AOS
Manage AOS device profiles.
This does not apply the resource to a rack type, template or
existing blueprint. See aos.rack_type
, aos.template
or aos.blueprint
to apply to the respective resource.
Source code in aos/devices.py
class AosDeviceProfiles(AosSubsystem):
"""
Manage AOS device profiles.
This does not apply the resource to a rack type, template or
existing blueprint. See `aos.rack_type`, `aos.template` or `aos.blueprint`
to apply to the respective resource.
"""
def get_all(self):
"""
Return all device profiles configured from AOS
Returns
-------
(obj) json response
"""
dp_path = "/api/device-profiles"
return self.rest.json_resp_get(dp_path)
def get_device_profile(self, dp_id: str = None, dp_name: str = None):
"""
Return an existing rack type by id or name
Parameters
----------
dp_id
(str) ID of AOS external router (optional)
dp_name
(str) Name or label of AOS external router (optional)
Returns
-------
(obj) json response
"""
if dp_name:
dev_profs = self.get_all()
if dev_profs:
for dp in dev_profs:
if dp.get("display_name") == dp_name:
return dp
raise AosAPIError(f"External Router {dp_name} not found")
return self.rest.json_resp_get(f"/api/device-profiles/{dp_id}")
def add_device_profiles(self, dp_list):
"""
Add one or more device profiles to AOS
Parameters
----------
dp_list
(list) - list of json payloads
Returns
-------
(list) device profile IDs
"""
p_path = "/api/device-profiles"
ids = []
i = 0
while i < len(dp_list):
resp = self.rest.json_resp_post(uri=p_path, data=dp_list[i])
if resp:
ids.append(resp["id"])
i += 1
if i % 30 == 0:
time.sleep(3)
return ids
def delete_device_profiles(self, dp_list: str):
"""
Delete one or more device profiles from AOS
Parameters
----------
dp_list
(list) - list of ids
Returns
-------
(list) deleted IDs
"""
p_path = "/api/device-profiles"
ids = []
for dp_id in dp_list:
self.rest.delete(f"{p_path}/{dp_id}")
ids.append(dp_id)
return ids
add_device_profiles(self, dp_list)
Add one or more device profiles to AOS
Parameters
dp_list (list) - list of json payloads
Returns
(list) device profile IDs
Source code in aos/devices.py
def add_device_profiles(self, dp_list):
"""
Add one or more device profiles to AOS
Parameters
----------
dp_list
(list) - list of json payloads
Returns
-------
(list) device profile IDs
"""
p_path = "/api/device-profiles"
ids = []
i = 0
while i < len(dp_list):
resp = self.rest.json_resp_post(uri=p_path, data=dp_list[i])
if resp:
ids.append(resp["id"])
i += 1
if i % 30 == 0:
time.sleep(3)
return ids
delete_device_profiles(self, dp_list)
Delete one or more device profiles from AOS
Parameters
dp_list (list) - list of ids
Returns
(list) deleted IDs
Source code in aos/devices.py
def delete_device_profiles(self, dp_list: str):
"""
Delete one or more device profiles from AOS
Parameters
----------
dp_list
(list) - list of ids
Returns
-------
(list) deleted IDs
"""
p_path = "/api/device-profiles"
ids = []
for dp_id in dp_list:
self.rest.delete(f"{p_path}/{dp_id}")
ids.append(dp_id)
return ids
get_all(self)
get_device_profile(self, dp_id=None, dp_name=None)
Return an existing rack type by id or name Parameters
dp_id (str) ID of AOS external router (optional) dp_name (str) Name or label of AOS external router (optional)
Returns
(obj) json response
Source code in aos/devices.py
def get_device_profile(self, dp_id: str = None, dp_name: str = None):
"""
Return an existing rack type by id or name
Parameters
----------
dp_id
(str) ID of AOS external router (optional)
dp_name
(str) Name or label of AOS external router (optional)
Returns
-------
(obj) json response
"""
if dp_name:
dev_profs = self.get_all()
if dev_profs:
for dp in dev_profs:
if dp.get("display_name") == dp_name:
return dp
raise AosAPIError(f"External Router {dp_name} not found")
return self.rest.json_resp_get(f"/api/device-profiles/{dp_id}")