加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
cmdb_inventory_plugin_v1.py 3.04 KB
一键复制 编辑 原始数据 按行查看 历史
mobiledj 提交于 2022-05-27 22:21 . init
# import logging
from typing import Any, Dict, Type, List
from nornir.core.inventory import (
Inventory,
Group,
Groups,
Host,
Hosts,
Defaults,
ConnectionOptions,
HostOrGroup,
ParentGroups,
)
# logger = logging.getLogger(__name__)
def _get_connection_options(data: Dict[str, Any]) -> Dict[str, ConnectionOptions]:
cp = {}
for cn, c in data.items():
cp[cn] = ConnectionOptions(
hostname=c.get("hostname"),
port=c.get("port"),
username=c.get("username"),
password=c.get("password"),
platform=c.get("platform"),
extras=c.get("extras"),
)
return cp
def _get_defaults(data: Dict[str, Any]) -> Defaults:
return Defaults(
hostname=data.get("hostname"),
port=data.get("port"),
username=data.get("username"),
password=data.get("password"),
platform=data.get("platform"),
data=data.get("data"),
connection_options=_get_connection_options(data.get("connection_options", {})),
)
def _get_inventory_element(
typ: Type[HostOrGroup], data: Dict[str, Any], name: str, defaults: Defaults
) -> HostOrGroup:
return typ(
name=name,
hostname=data.get("hostname") or data.get('ip'),
port=data.get("port"),
username=data.get("username"),
password=data.get("password"),
platform=data.get("platform"),
data=data.get("data"),
groups=data.get(
"groups"
), # this is a hack, we will convert it later to the correct type
defaults=defaults,
connection_options=_get_connection_options(data.get("connection_options", {})),
)
class CMDBInventory:
def __init__(
self,
devices
) -> None:
"""
:param devices:
"""
host_info = ['hostname', 'username', 'password', 'port', 'platform', 'ip', 'device_type']
reshape_devices = []
for device in devices:
reshape_device = {
'data': {},
}
for k, v in device.items():
if k in host_info:
if k == 'device_type':
reshape_device['platform'] = v
else:
reshape_device[k] = v
else:
reshape_device['data'][k] = v
reshape_device['connection_options'] = {
'netmiko': {'extras': {'global_delay_factor': device.get('global_delay_factor'),
'secret': device.get('secret')}}}
reshape_devices.append(reshape_device)
self.devices = reshape_devices
def load(self) -> Inventory:
defaults = Defaults()
groups = Groups()
hosts = Hosts()
hosts_dict = {device['ip']: device for device in self.devices}
for name, hosts_dict in hosts_dict.items():
hosts[name] = _get_inventory_element(Host, hosts_dict, name, defaults)
return Inventory(hosts=hosts, groups=groups, defaults=defaults)
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化