diff --git a/.github/workflows/Pynetbox.yaml b/.github/workflows/Pynetbox.yaml index 84f45a5d..694ead82 100644 --- a/.github/workflows/Pynetbox.yaml +++ b/.github/workflows/Pynetbox.yaml @@ -35,3 +35,11 @@ jobs: run: | cd Pynetbox_Data_Uploader && pylint $(git ls-files '*.py') --rcfile=.pylintrc + - name: Run tests and collect coverage + run: cd Pynetbox_Data_Uploader && python3 -m pytest --cov-report xml:coverage.xml --cov + + - name: Upload coverage to Codecov + uses: codecov/codecov-action@v3 + with: + files: ./Pynetbox_Data_Uploader/coverage.xml + diff --git a/Pynetbox_Data_Uploader/.pylintrc b/Pynetbox_Data_Uploader/.pylintrc index f7ce6a30..91b8da9e 100644 --- a/Pynetbox_Data_Uploader/.pylintrc +++ b/Pynetbox_Data_Uploader/.pylintrc @@ -7,5 +7,4 @@ max-line-length=120 # C0114: Missing module string - we don't need module strings for the small repo # W0511: TODOs they're well....to do later - disable=C0114,W0511 \ No newline at end of file diff --git a/Pynetbox_Data_Uploader/lib/netbox_api/netbox_check.py b/Pynetbox_Data_Uploader/lib/netbox_api/netbox_check.py index 20335782..b303105d 100644 --- a/Pynetbox_Data_Uploader/lib/netbox_api/netbox_check.py +++ b/Pynetbox_Data_Uploader/lib/netbox_api/netbox_check.py @@ -3,8 +3,8 @@ class NetboxCheck: This class contains methods that check if an object exists in Netbox. """ - def __init__(self, netbox): - self.netbox = netbox + def __init__(self, api): + self.netbox = api def check_device_exists(self, device_name: str) -> bool: """ diff --git a/Pynetbox_Data_Uploader/lib/netbox_api/netbox_create.py b/Pynetbox_Data_Uploader/lib/netbox_api/netbox_create.py index 8aba1d63..95ac241c 100644 --- a/Pynetbox_Data_Uploader/lib/netbox_api/netbox_create.py +++ b/Pynetbox_Data_Uploader/lib/netbox_api/netbox_create.py @@ -6,8 +6,8 @@ class NetboxCreate: This class contains methods that will interact create objects in Netbox. """ - def __init__(self, netbox): - self.netbox = netbox + def __init__(self, api): + self.netbox = api def create_device(self, data: Union[Dict, List]) -> bool: """ diff --git a/Pynetbox_Data_Uploader/lib/netbox_api/netbox_data.py b/Pynetbox_Data_Uploader/lib/netbox_api/netbox_data.py index 4236c923..1f94fc1d 100644 --- a/Pynetbox_Data_Uploader/lib/netbox_api/netbox_data.py +++ b/Pynetbox_Data_Uploader/lib/netbox_api/netbox_data.py @@ -11,11 +11,11 @@ class NetboxGetID: This class retrieves field value ID's from Netbox. """ - def __init__(self, netbox): + def __init__(self, api): """ This method allows the Netbox Api Object and Enums to be accessible within the class. """ - self.netbox = netbox + self.netbox = api self.enums_id = DeviceInfoID self.enums_no_id = DeviceInfoNoID @@ -39,6 +39,9 @@ def get_id( site_name = self.netbox.dcim.sites.get(site_value).name site_slug = site_name.replace(" ", "-").lower() value = value.get(name=netbox_value, site=site_slug) + list_value = list(value) + list_value = [item for item in list_value if item[0] == "id"] + value = list_value[0][1] else: value = value.get(name=netbox_value).id return value @@ -50,7 +53,7 @@ def get_id_from_key(self, key: str, dictionary: Dict) -> Union[str, int]: :param dictionary: The device dictionary being referenced. :return: If an ID was needed and found it returns the ID. If an ID was not needed it returns the original value. """ - if key not in list(self.enums_no_id.__members__): + if key.upper() not in list(self.enums_no_id.__members__): value = self.get_id( attr_string=key, netbox_value=dictionary[key], diff --git a/Pynetbox_Data_Uploader/lib/user/__init__.py b/Pynetbox_Data_Uploader/lib/user_methods/__init__.py similarity index 100% rename from Pynetbox_Data_Uploader/lib/user/__init__.py rename to Pynetbox_Data_Uploader/lib/user_methods/__init__.py diff --git a/Pynetbox_Data_Uploader/lib/user_methods/csv_to_netbox.py b/Pynetbox_Data_Uploader/lib/user_methods/csv_to_netbox.py new file mode 100644 index 00000000..8def1b2a --- /dev/null +++ b/Pynetbox_Data_Uploader/lib/user_methods/csv_to_netbox.py @@ -0,0 +1,116 @@ +from typing import List +import argparse +from lib.netbox_api.netbox_create import NetboxCreate +from lib.netbox_api.netbox_connect import NetboxConnect +from lib.netbox_api.netbox_check import NetboxCheck +from lib.utils.format_dict import FormatDict + +# pylint:disable = broad-exception-raised +# Disabled this pylint warning as the exception doesn't catch an error. +# We want it to stop the program if a device already exists in netbox. + + +class CsvToNetbox: + """ + This class contains organised methods in the 4 step proccess of reading csv's to then uploading to Netbox. + """ + def __init__(self, url: str, token: str): + """ + This initialises the class with the following parameters. + It also allows the rest of the class to access the imported Classes. + :param url: The Netbox url. + :param token: The Netbox auth token. + """ + self.netbox = NetboxConnect(url, token).api_object() + self.format_dict = FormatDict(self.netbox) + self.exist = NetboxCheck(self.netbox) + self.create = NetboxCreate(self.netbox) + + def read_csv(self, file_path) -> List: + """ + This method calls the csv_to_python and seperate_data method. + This will take the csv file and return a list of device dictionaries. + :param file_path: The file path to the csv file to be read. + :return: Returns a list of devices + """ + print("Reading CSV...") + device_data = self.format_dict.csv_to_python(file_path) + device_list = self.format_dict.separate_data(device_data) + print("Read CSV.") + return device_list + + def check_netbox(self, device_list: List) -> bool: + """ + This method calls the check_device_exists and check_device_type_exists method on each device in the list. + :param device_list: A list of devices. + :return: Returns True if the devices don't exist and device types do exist. Raises an Exception otherwise. + """ + print("Checking devices in Netbox...") + for device in device_list: + device_exist = self.exist.check_device_exists(device["name"]) + if device_exist: + raise Exception(f'Device {device["name"]} already exists in Netbox.') + type_exist = self.exist.check_device_type_exists(device["device_type"]) + if not type_exist: + raise Exception(f'Type {device["device_type"]} does not exist.') + print("Checked devices.") + return True + + def convert_data(self, device_list: List) -> List: + """ + This method calls the iterate_dict method. + :param device_list: A list of devices. + :return: Returns the updated list of devices. + """ + print("Formatting data...") + formatted_list = self.format_dict.iterate_dicts(device_list) + print("Formatted data.") + return formatted_list + + def send_data(self, device_list: List) -> bool: + """ + This method calls the device create method to create devices in Netbox. + :param device_list: A list of devices. + :return: Returns bool whether the devices where created. + """ + print("Sending data to Netbox...") + devices = self.create.create_device(device_list) + print("Sent data.") + return bool(devices) + + +def arg_parser(): + """ + This function creates a parser object and adds 3 arguments to it. + This allows users to run the python file with arguments. Like a script. + """ + parser = argparse.ArgumentParser( + description="Create devices in Netbox from CSV files.", + usage="python csv_to_netbox.py url token file_path", + ) + parser.add_argument("url", help="The Netbox URL.") + parser.add_argument("token", help="Your Netbox Token.") + parser.add_argument("file_path", help="Your file path to csv files.") + return parser.parse_args() + + +def do_csv_to_netbox(args) -> bool: + """ + This function calls the methods from CsvToNetbox class. + :param args: The arguments from argparse. Supplied when the user runs the file from CLI. + :return: Returns bool if devices where created or not. + """ + class_object = CsvToNetbox(url=args.url, token=args.token) + device_list = class_object.read_csv(args.file_path) + class_object.check_netbox(device_list) + format_list = class_object.convert_data(device_list) + result = class_object.send_data(format_list) + return result + + +if __name__ == "__main__": + arguments = arg_parser() + if do_csv_to_netbox(arguments): + print("Done.") + else: + print("Uh Oh.") diff --git a/Pynetbox_Data_Uploader/lib/utils/format_dict.py b/Pynetbox_Data_Uploader/lib/utils/format_dict.py index d66eed74..55f934f7 100644 --- a/Pynetbox_Data_Uploader/lib/utils/format_dict.py +++ b/Pynetbox_Data_Uploader/lib/utils/format_dict.py @@ -1,66 +1,67 @@ -from typing import Dict, List -from pandas import read_csv -from lib.netbox_api.netbox_data import NetboxGetID - - -class FormatDict: - """ - This class takes dictionaries with string values and changes those to ID values from Netbox. - """ - - def __init__(self, netbox): - """ - This method initialises the class with the following parameters. - :param netbox: The Netbox object to pass into NetboxGetID - """ - self.netbox = netbox - - def iterate_dicts(self, dicts: list) -> List: - """ - This method iterates through each dictionary and calls a format method on each. - :return: Returns the formatted dictionaries. - """ - new_dicts = [] - for dictionary in dicts: - new_dicts.append(self.format_dict(dictionary)) - return new_dicts - - def format_dict(self, dictionary) -> Dict: - """ - This method iterates through each value in the dictionary. - If the value needs to be converted into a Pynetbox ID it calls the .get() method. - :param dictionary: The dictionary to be formatted - :return: Returns the formatted dictionary - """ - for key in dictionary: - netbox_id = NetboxGetID(self.netbox).get_id_from_key(key=key, dictionary=dictionary) - dictionary[key] = netbox_id - return dictionary - - - @staticmethod - def csv_to_python(file_path: str) -> Dict: - """ - This method reads data from csv files and writes them to a dictionary. - :param file_path: The file path of the utils file to be read from. - :return: Returns the data from the csv as a dictionary. - """ - dataframe = read_csv(file_path) - return dataframe.to_dict(orient="list") - - @staticmethod - def separate_data(data: dict) -> List: - """ - This method reduces Pandas utils to Dict conversion to individual dictionaries. - :param data: The data from the utils file. - :return: Returns a list of dictionaries which each represent a row of data from utils. - """ - data_keys = list(data.keys()) - len_rows = len(data[data_keys[0]]) - dicts = [] - for index in range(len_rows): - new_dict = {} - for key in data_keys: - new_dict.update({key: data[key][index]}) - dicts.append(new_dict) - return dicts +from typing import Dict, List +from pandas import read_csv +from lib.netbox_api.netbox_data import NetboxGetID + + +class FormatDict: + """ + This class takes dictionaries with string values and changes those to ID values from Netbox. + """ + + def __init__(self, api): + """ + This method initialises the class with the following parameters. + :param api: The Netbox object to pass into NetboxGetID + """ + self.netbox = api + + def iterate_dicts(self, dicts: list) -> List: + """ + This method iterates through each dictionary and calls a format method on each. + :param dicts: A list of dictionaries to be formatted. + :return: Returns the formatted dictionaries. + """ + new_dicts = [] + for dictionary in dicts: + new_dicts.append(self.format_dict(dictionary)) + return new_dicts + + def format_dict(self, dictionary) -> Dict: + """ + This method iterates through each value in the dictionary. + If the value needs to be converted into a Pynetbox ID it calls the .get() method. + :param dictionary: The dictionary to be formatted + :return: Returns the formatted dictionary + """ + for key in dictionary: + netbox_id = NetboxGetID(self.netbox).get_id_from_key(key=key, dictionary=dictionary) + dictionary[key] = netbox_id + return dictionary + + + @staticmethod + def csv_to_python(file_path: str) -> Dict: + """ + This method reads data from csv files and writes them to a dictionary. + :param file_path: The file path of the utils file to be read from. + :return: Returns the data from the csv as a dictionary. + """ + dataframe = read_csv(file_path) + return dataframe.to_dict(orient="list") + + @staticmethod + def separate_data(data: dict) -> List: + """ + This method reduces Pandas utils to Dict conversion to individual dictionaries. + :param data: The data from the utils file. + :return: Returns a list of dictionaries which each represent a row of data from utils. + """ + data_keys = list(data.keys()) + len_rows = len(data[data_keys[0]]) + dicts = [] + for index in range(len_rows): + new_dict = {} + for key in data_keys: + new_dict.update({key: data[key][index]}) + dicts.append(new_dict) + return dicts diff --git a/Pynetbox_Data_Uploader/pytest.ini b/Pynetbox_Data_Uploader/pytest.ini index 65c35ff4..f063b7c5 100644 --- a/Pynetbox_Data_Uploader/pytest.ini +++ b/Pynetbox_Data_Uploader/pytest.ini @@ -2,4 +2,5 @@ pythonpath = lib testpaths = Tests python_files = *.py -python_functions = test_* \ No newline at end of file +python_functions = test_* +addopts = --ignore=setup.py \ No newline at end of file diff --git a/Pynetbox_Data_Uploader/requirements.txt b/Pynetbox_Data_Uploader/requirements.txt index a12310d5..687a9166 100644 Binary files a/Pynetbox_Data_Uploader/requirements.txt and b/Pynetbox_Data_Uploader/requirements.txt differ diff --git a/Pynetbox_Data_Uploader/setup.py b/Pynetbox_Data_Uploader/setup.py new file mode 100644 index 00000000..f1c479f9 --- /dev/null +++ b/Pynetbox_Data_Uploader/setup.py @@ -0,0 +1,20 @@ +from setuptools import setup, find_packages + +VERSION = "0.1.0" +DESCRIPTION = "python package for PYNETBOX tools" + +LONG_DESCRIPTION = """Python package to interact with Netbox from cli.""" + +setup( + name="Pynetbox_Data_Uploader", + version=VERSION, + author="Kalibh Halford", + author_email="", + description=DESCRIPTION, + long_description=LONG_DESCRIPTION, + packages=find_packages(), + package_dir={"Pynetbox_Data_Uploader": "lib"}, + python_requires=">=3.9", + install_requires=[], + keywords=["python"], +)