Skip to content

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

Source code in aos/devices.py
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"]
    ]

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)

Return all device profiles configured from AOS

Returns

(obj) json response
Source code in aos/devices.py
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)

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}")