from first import first
from jnpr.healthbot.exception import SchemaError, NotFoundError
from jnpr.healthbot.swagger.models.device_schema import DeviceSchema
from jnpr.healthbot.swagger.models.device_group_schema import DeviceGroupSchema
from jnpr.healthbot.swagger.models.network_group_schema import NetworkGroupSchema
from jnpr.healthbot.swagger.models.device_health_tree import DeviceHealthTree
from jnpr.healthbot.swagger.models.device_group_health_tree import DeviceGroupHealthTree
from jnpr.healthbot.swagger.models.network_health_tree import NetworkHealthTree
from jnpr.healthbot.modules import BaseModule
import logging
logger = logging.getLogger(__file__)
[docs]class Device(BaseModule):
[docs] def __init__(self, hbot):
"""
:param object hbot: :class:`jnpr.healthbot.HealthBotClient` client instance
"""
super().__init__(hbot)
[docs] def add(self, device_id: str = None, host: str = None,
username: str = None, password: str = None,
schema: DeviceSchema = None, **kwargs):
"""
Add device to HealthBot
:param str device_id: The name of the device as provided by the User
:param str host: The hostname/ip-address of the target device
:param str username: The login user-name for the target device
:param str password: The login password for the user
:param object schema: `DeviceSchema <jnpr.healthbot.swagger.models.html#deviceschema>`_
Example:
::
from jnpr.healthbot import HealthBotClient
from jnpr.healthbot import DeviceSchema
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
ds = DeviceSchema(device_id='xyz', host='xx.xxx.xxx.xxx',
authentication={"password": {"password": "xxxxx", "username": "xxxxx"}})
# we can also later assign values like this
ds.description = "HbEZ testing"
# This will add device in candidate DB
hb.device.add(schema=ds)
# commit changes to master DB
hb.commit()
"""
def _add_device(device_id=None):
if kwargs:
if schema is not None:
raise SystemError("schema and kwargs are mutually exclusive")
device_schema = DeviceSchema(device_id=device_id, host=host, **kwargs)
if username is not None and password is not None:
device_schema.authentication = {
"password": {
"password": password,
"username": username
}
}
payload = self.hbot._create_payload(device_schema)
elif schema is not None:
payload = self.hbot._create_payload(schema)
device_id = schema.device_id
else:
payload = {
"authentication": {
"password": {
"password": password,
"username": username
}
},
"device-id": device_id,
"host": host,
}
url = self.hbot.urlfor.device(device_id)
response = self.api.post(url, json=payload)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return True
if schema is not None:
if not isinstance(schema, DeviceSchema):
raise SchemaError(DeviceSchema)
device_id = schema.device_id
host = schema.host
devices_list_url = self.hbot.urlfor.devices() + self.hbot.apiopt_candidate
resp = self.api.get(devices_list_url)
# The API will return a 404 if there are no devices in the system. We need to check for
# this condition
if resp.status_code == 404:
return _add_device(device_id)
# examine the existing devices and see if there is one that already
# exists by this name and host values
existing_devices = resp.json()['device']
found = first(
filter(
lambda i: i['device-id'] == device_id and i['host'] == host,
existing_devices))
if found:
logger.debug(
"Device with given device-id '{}' already exists. Updating same.".format(device_id))
# if we are here, then we need to add this new device
return _add_device(device_id)
[docs] def delete(self, device_id: str, force: bool = False):
"""
Remove device from HealthBot
:param str device_id: The name of the device as provided by the User
:param bool force: If True, Delete given device from all the device
group (if present)
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
# This will delete device in candidate DB
hb.device.delete('xyz')
# commit changes to master DB
hb.commit()
:returns: True when OK
"""
device_url = self.hbot.urlfor.device(device_id)
if force:
groups_list_url = self.hbot.urlfor.device_groups()
# Remove Device from the Device Group (requirement before removing
# Device)
device_groups_response = self.api.get(groups_list_url)
# device_groups is the list of dictionaries, each containing the information of
# device groups present in HBot
device_groups = device_groups_response.json()
for each_device_group in device_groups["device-group"]:
device_list = each_device_group['devices']
device_group_name = each_device_group['device-group-name']
# checking if the given device_id present in this device group
if device_id in device_list:
# Delete the device from the device group
device_list.remove(device_id)
dev_grp_schema = self.hbot.device_group.get(
device_group_name)
dev_grp_schema.devices = device_list
self.hbot.device_group.update(dev_grp_schema)
# Remove Device from HealthBot
response = self.api.delete(device_url)
if response.status_code != 204:
logger.error(response.text)
response.raise_for_status()
return True
[docs] def get_ids(self):
"""
Return Device IDs for all the devices in HealthBot system
:return: list of device IDs
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
print(hb.device.get_ids())
"""
devices_list_url = self.hbot.urlfor.device()
resp = self.api.get(devices_list_url)
if resp.status_code == 404:
return []
return resp.json()
[docs] def get(self, device_id: str = None, uncommitted: bool = True):
"""
Get `DeviceSchema <jnpr.healthbot.swagger.models.html#deviceschema>`_ for
given device id or list for all devices
:param str device_id: The name of the device as provided by the User
:param bool uncommitted: True includes fetches uncommitted changes,
False restricts data set to only committed changes
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
device = hb.device.get('vmx')
print(device)
devices = hb.device.get()
for device in devices:
print(device)
:return: `DeviceSchema(s) <jnpr.healthbot.swagger.models.html#deviceschema>`_
"""
if device_id is not None:
device_url = self.hbot.urlfor.device(device_id)
if uncommitted:
device_url += self.hbot.apiopt_candidate
response = self.api.get(device_url)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return self.hbot._create_schema(response, DeviceSchema)
else:
device_list = []
devices_list_url = self.hbot.urlfor.devices()
if uncommitted:
devices_list_url += self.hbot.apiopt_candidate
resp = self.api.get(devices_list_url)
if resp.status_code == 404:
return []
# examine the existing devices and return a list of devices
existing_devices = resp.json()['device']
for device in existing_devices:
device = self.hbot._create_schema(device, DeviceSchema)
device_list.append(device)
return device_list
[docs] def update(self, schema: DeviceSchema = None, **kwargs):
"""
Update `DeviceSchema <jnpr.healthbot.swagger.models.html#deviceschema>`_ for
given device schema
Passing Schema invoke `put` and kwargs `post`
:param obj schema: `DeviceSchema <jnpr.healthbot.swagger.models.html#deviceschema>`_
:param object kwargs: key values, which can be used to create
DeviceSchema
Check `DeviceSchema <jnpr.healthbot.swagger.models.html#deviceschema>`_
for details about which all keys can be used
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
schemaObj = hb.device.get('xyz')
schemaObj.description = 'changed description'
hb.device.update(schemaObj)
hb.device.update(device_id="xyz", host='xx.xxx.x.xx', system_id="xxxx")
:returns: True when OK
"""
if schema is None:
schema = DeviceSchema(**kwargs)
call = self.api.post
else:
if not isinstance(schema, DeviceSchema):
raise SchemaError(DeviceSchema)
call = self.api.put
payload = self.hbot._create_payload(schema)
device_id = schema.device_id
device_url = self.hbot.urlfor.device(device_id)
response = call(device_url, json=payload)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return True
[docs] def get_facts(self, device_id: str = None, uncommitted: bool = True):
"""
Get device(s) facts. Get facts for provided device id, if device
id is not provided, get facts for all devices
:param str device_id: The name of the device as provided by the User
:param bool uncommitted: True includes fetches uncommitted changes,
False restricts data set to only committed changes
Example:
::
from jnpr.healthbot import HealthBotClient
from pprint import pprint
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
facts = hb.device.get_facts('vmx')
pprint(facts)
facts = hb.device.get_facts()
pprint(facts)
:return: Single/List of dicts of facts
"""
if device_id is not None:
device_url = self.hbot.urlfor.device_facts(device_id)
if uncommitted:
device_url += self.hbot.apiopt_candidate
response = self.api.get(device_url)
response.raise_for_status()
return response.json()
else:
device_url = self.hbot.urlfor.devices_facts()
if uncommitted:
device_url += self.hbot.apiopt_candidate
response = self.api.get(device_url)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return response.json()
[docs] def health(self, device_id: str):
"""
Returns health of given Device id
`DeviceHealthTree <jnpr.healthbot.swagger.models.html#deviceheathtree>`_
:param str device_id: The name of the device as provided by the User
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
print(hb.device.health('core'))
:return: `DeviceHealthTree <jnpr.healthbot.swagger.models.html#deviceheathtree>`_
"""
device_health_url = self.hbot.urlfor.device_health(device_id)
resp = self.api.get(device_health_url)
if resp.status_code == 404:
return {}
return self.hbot._create_schema(resp, DeviceHealthTree)
[docs]class DeviceGroup(BaseModule):
[docs] def __init__(self, hbot):
"""
:param object hbot: :class:`jnpr.healthbot.HealthBotClient` client instance
"""
super().__init__(hbot)
[docs] def add(self, schema: DeviceGroupSchema = None, **kwargs):
"""
Add device group to HealthBot
:param object schema: `DeviceGroupSchema <jnpr.healthbot.swagger.models.html#devicegroupschema>`_
:param object kwargs: key values, which can be used to create
DeviceGroupSchema
Check `DeviceGroupSchema <jnpr.healthbot.swagger.models.html#devicegroupschema>`_
for details about which all keys can be used
Example:
::
from jnpr.healthbot import HealthBotClient
from jnpr.healthbot import DeviceSchema
from jnpr.healthbot import DeviceGroupSchema
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
ds = DeviceSchema(device_id='xyz', host='xx.xxx.xxx.xxx',
authentication={"password": {"password": "xxxxx", "username": "xxxxx"}})
# This will add device in candidate DB
hb.device.add(schema=ds)
dgs = DeviceGroupSchema(device_group_name="edge",
description="All devices on the edge",
devices=['xyz'])
hb.device_group.add(dgs)
# commit changes to master DB
hb.commit()
:returns: True when OK
"""
if schema is None:
schema = DeviceGroupSchema(**kwargs)
else:
if not isinstance(schema, DeviceGroupSchema):
raise SchemaError(DeviceGroupSchema)
payload = self.hbot._create_payload(schema)
device_group_url = self.hbot.urlfor.device_group(
payload['device-group-name'])
# Add Device Group to the Healthbot
response = self.api.post(device_group_url, json=payload)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return True
[docs] def delete(self, device_group_name: str, force: bool = False):
"""
Remove device group to HealthBot
:param str device_group_name: The name of the device group to be deleted
:param bool force: If True, First delete services for given device group
Example:
::
hb.devices.delete('edge')
hb.commit()
:returns: True when OK
"""
if force:
services_device_group_url = self.hbot.urlfor.services_device_group()
response = self.api.get(services_device_group_url)
if device_group_name in response.json():
services_device_group_url = self.hbot.urlfor.services_device_group(
device_group_name)
response = self.api.delete(services_device_group_url)
if response.status_code != 204:
logger.error(response.text)
response.raise_for_status()
device_group_url = self.hbot.urlfor.device_group(device_group_name)
response = self.api.delete(device_group_url)
if response.status_code != 204:
logger.error(response.text)
response.raise_for_status()
return True
[docs] def get(self, device_group_name: str = None, uncommitted: bool = True):
"""
Get `DeviceGroupSchema <jnpr.healthbot.swagger.models.html#devicegroupschema>`_ for
given device group name or list for all device groups
:param str device_group_name: Name of the device-group
:param bool uncommitted: True includes fetches uncommitted changes,
False restricts data set to only committed changes
Example:
::
device_group_schema = hb.device_group.get('edge')
groups = hb.device_group.get()
for group in groups:
print(group)
:return: `DeviceGroupSchema(s) <jnpr.healthbot.swagger.models.html#devicegroupschema>`_
"""
if device_group_name is not None:
device_group_url = self.hbot.urlfor.device_group(device_group_name)
if uncommitted:
device_group_url += self.hbot.apiopt_candidate
res = self.api.get(device_group_url)
# if 404 is returned, it means there is not a device group by that name
# we need to explicitly check for this given the API behavior
if res.status_code == 404:
return None
return self.hbot._create_schema(res, DeviceGroupSchema)
else:
device_groups_url = self.hbot.urlfor.device_groups()
if uncommitted:
device_groups_url += self.hbot.apiopt_candidate
response = self.api.get(device_groups_url)
if response.status_code == 404:
return []
response_json = response.json()
if response.status_code != 200:
logger.error(response.text)
raise NotFoundError(response_json)
device_group_list = []
existing_device_groups = response_json['device-group']
for device_group in existing_device_groups:
obj = self.hbot._create_schema(device_group, DeviceGroupSchema)
device_group_list.append(obj)
return device_group_list
[docs] def update(self, schema: DeviceGroupSchema = None, **kwargs):
"""
Update `DeviceGroupSchema <jnpr.healthbot.swagger.models.html#devicegroupschema>`_ for
given device schema
Passing Schema invoke `put` and kwargs `post`
:param obj schema: `DeviceGroupSchema <jnpr.healthbot.swagger.models.html#devicegroupschema>`_
:param object kwargs: key values, which can be used to create
DeviceGroupSchema
Check `DeviceGroupSchema <jnpr.healthbot.swagger.models.html#devicegroupschema>`_
for details about which all keys can be used
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
schemaObj = hb.device_group.get('Core')
schemaObj.description = "Changed"
hb.device_group.update(schemaObj)
:returns: True when OK
"""
if schema is None:
schema = DeviceGroupSchema(**kwargs)
call = self.api.post
else:
if not isinstance(schema, DeviceGroupSchema):
raise SchemaError(DeviceGroupSchema)
call = self.api.put
payload = self.hbot._create_payload(schema)
device_group_url = self.hbot.urlfor.device_group(
payload['device-group-name'])
response = call(device_group_url, json=payload)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return True
[docs] def check_device_in_group(self, device_name: str, device_group_name: str):
"""
This method check if the device is a member of the given device-group
:param str device_name: Name of the device
:param str device_group_name: Name of the device-group
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
print(hb.device_group.check_device_in_group('vmx', 'QFabric'))
Returns:
True if action successful
"""
# retrieve the current device-group dataset and determine if the given device name is already
# a member of the group. If so, return True
group_data = self.get(device_group_name)
if group_data is None:
return False
device_list = group_data.to_dict().get('devices', [])
return device_name in device_list
[docs] def add_device_in_group(self, device_name: str, device_group_name: str):
"""
This method ensures that the given device is a member of the given
device-group
:param str device_name: Name of the device
:param str device_group_name: Name of the device-group
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
hb.device_group.add_device_in_group('vmx', 'QFabric')
Raises:
HTTPError: When error making changes via the HBOT API
Returns:
True if action successful
"""
# retrieve the current device-group dataset and determine if the given device name is already
# a member of the group. If so, return True
if self.check_device_in_group(device_name, device_group_name):
return True
# add the device to the list and then store back to the server
group_data = self.get(device_group_name)
if group_data is None:
device_list = []
group_data = DeviceGroupSchema(device_group_name=device_group_name)
else:
device_list = group_data.to_dict().get('devices', [])
device_list.append(device_name)
group_data.devices = device_list
return self.add(group_data)
[docs] def health(self, device_group_name: str):
"""
Returns health of given Device id
`DeviceGroupHealthTree <jnpr.healthbot.swagger.models.html#devicegroupheathtree>`_
:param str device_group_name: The name of the device group
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
print(hb.device_group.health('edge'))
:return: `DeviceGroupHealthTree <jnpr.healthbot.swagger.models.html#devicegroupheathtree>`_
"""
device_group_health_url = self.hbot.urlfor.device_group_health(
device_group_name)
resp = self.api.get(device_group_health_url)
if resp.status_code == 404:
return {}
return self.hbot._create_schema(resp, DeviceGroupHealthTree)
[docs]class NetworkGroup(BaseModule):
[docs] def __init__(self, hbot):
"""
:param object hbot: :class:`jnpr.healthbot.HealthBotClient` client instance
"""
super().__init__(hbot)
[docs] def add(self, schema: NetworkGroupSchema = None, **kwargs):
"""
Create Network Group
:param object schema: `NetworkGroupSchema <jnpr.healthbot.swagger.models.html#networkgroupschema>`_
:param object kwargs: key values, which can be used to create
NetworkGroupSchema
Check `NetworkGroupSchema <jnpr.healthbot.swagger.models.html#networkgroupschema>`_
for details about which all keys can be used
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
hb.devices.add_network_group(network_group_name="HbEZ")
# or
from jnpr.healthbot import HealthBotClient
from jnpr.healthbot import NetworkGroupSchema
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
ngs = NetworkGroupSchema(network_group_name="HbEZ")
hb.network_group.add(schema = ngs)
"""
if schema is None:
schema = NetworkGroupSchema(**kwargs)
else:
if not isinstance(schema, NetworkGroupSchema):
raise SchemaError(NetworkGroupSchema)
payload = self.hbot._create_payload(schema)
# check if the group already exists. If it does, then return now
# ... this means this function will not do an "ensure / merge" of data
# if the group already exists
network_group_url = self.hbot.urlfor.network_group(
payload['network-group-name'])
resp = self.api.get(network_group_url + self.hbot.apiopt_candidate)
if resp.ok:
return True
response = self.api.post(network_group_url, json=payload)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return True
[docs] def delete(self, network_group_name: str):
"""
Delete Network Group
:param str network_group_name: The name of the network group to be deleted
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
hb.network_group.delete(network_group_name="HbEZ")
"""
payload = {'network-group-name': network_group_name}
network_group_url = self.hbot.urlfor.network_group(network_group_name)
response = self.api.delete(network_group_url, json=payload)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return True
[docs] def get(self, network_group_name: str = None, uncommitted: bool = True):
"""
get Network Group(s) details
:param str network_group_name: The name of the network group to be fetched
:param bool uncommitted: True includes fetches uncommitted changes,
False restricts data set to only committed changes
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
print(hb.network_group.get(network_group_name="HbEZ"))
# for all network groups
print(hb.network_group.get())
"""
if network_group_name is not None:
network_group_url = self.hbot.urlfor.network_group(
network_group_name)
if uncommitted:
network_group_url += self.hbot.apiopt_candidate
response = self.api.get(network_group_url)
if response.status_code != 200:
logger.error(response.text)
response.raise_for_status()
return self.hbot._create_schema(response, NetworkGroupSchema)
else:
network_groups_url = self.hbot.urlfor.network_groups()
if uncommitted:
network_groups_url += self.hbot.apiopt_candidate
response = self.api.get(network_groups_url)
if response.status_code == 404:
return []
response_json = response.json()
if response.status_code != 200:
logger.error(response.text)
raise NotFoundError(response_json)
response.raise_for_status()
network_group_list = []
existing_network_groups = response_json['network-group']
for network_group in existing_network_groups:
obj = self.hbot._create_schema(
network_group, NetworkGroupSchema)
network_group_list.append(obj)
return network_group_list
[docs] def update(self, schema: NetworkGroupSchema = None, **kwargs):
"""
Update `NetworkGroupSchema <jnpr.healthbot.swagger.models.html#networkgroupschema>`_ for
given network schema object
Passing Schema invoke `put` and kwargs `post`
:param obj schema: `NetworkGroupSchema <jnpr.healthbot.swagger.models.html#networkgroupschema>`_
:param object kwargs: key values, which can be used to create
NetworkGroupSchema.
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
schemaObj = hb.network_group.get("HbEZ")
schemaObj.description = "HbEZ example"
hb.network_group.update(schemaObj)
:returns: True when OK
"""
if schema is None:
schema = NetworkGroupSchema(**kwargs)
call = self.api.post
else:
if not isinstance(schema, NetworkGroupSchema):
raise SchemaError(NetworkGroupSchema)
call = self.api.put
payload = self.hbot._create_payload(schema)
network_group_url = self.hbot.urlfor.network_group(
payload['network-group-name'])
response = call(network_group_url, json=payload)
if response.status_code != 200:
logger.error(response.text)
return True
[docs] def health(self, network_group_name: str):
"""
Returns health of given Device id
`NetworkHealthTree <jnpr.healthbot.swagger.models.html#networkheathtree>`_
:param str network_group_name: The name of the network group
Example:
::
from jnpr.healthbot import HealthBotClient
with HealthBotClient('xx.xxx.x.xx', 'xxxx', 'xxxx') as hb:
print(hb.network_group.health('core'))
:return: `NetworkHealthTree <jnpr.healthbot.swagger.models.html#networkheathtree>`_
"""
network_group_health_url = self.hbot.urlfor.network_group_health(
network_group_name)
resp = self.api.get(network_group_health_url)
if resp.status_code == 404:
return {}
return self.hbot._create_schema(resp, NetworkHealthTree)
