v1.0.7

.gitignore

@@ -1,3 +1,4 @@
+*.swp
 *.pyc
 .cache
 .idea

VERSION

@@ -1 +1 @@
-1.0.6
+1.0.7

netbox/netbox.py

@@ -20,6 +20,7 @@
 import yaml
 import argparse
 import requests
+
 try:
     import json
 except ImportError:
@@ -46,39 +47,6 @@ def cli_arguments():
 
 
 # Utils.
-def reduce_path(source_dict, key_path):
-    for key in key_path:
-        if isinstance(source_dict.get(key), dict) and len(key_path) > 1:
-            source_dict = source_dict.get(key)
-            key_path = key_path[1:]
-            reduce_path(source_dict, key_path)
-        else:
-            return source_dict[key]
-
-
-def get_value_by_path(source_dict, key_path, ignore_key_error=False):
-    """Get key value from nested dict by path.
-
-    Args:
-        source_dict: The dict that we look into.
-        key_path: The path of key in dot notion. e.g. "parent_dict.child_dict.key_name"
-        ignore_key_error: Ignore KeyError if the key not found in provided path.
-
-    Returns:
-        If key found in provided path, it will be returned.
-        If not, None will be returned.
-    """
-
-    try:
-        key_output = reduce_path(source_dict, key_path)
-    except KeyError as key_name:
-        if ignore_key_error:
-            key_output = None
-        else:
-            sys.exit("The key %s is not found. Please remember, Python is case sensitive." % key_name)
-    return key_output
-
-
 def open_yaml_file(yaml_file):
     """Open YAML file.
 
@@ -117,14 +85,10 @@ def __init__(self, script_args, script_config_data):
         self.host = script_args.host
 
         # Script configuration.
-        main_config_key = "netbox"
-        script_config = script_config_data.get(main_config_key)
-        if script_config:
-            self.api_url = script_config["main"].get('api_url')
-            self.group_by = script_config.setdefault("group_by", {})
-            self.hosts_vars = script_config.setdefault("hosts_vars", {})
-        else:
-            sys.exit("The key '%s' is not found in config file." % main_config_key)
+        self.script_config = script_config_data
+        self.api_url = self._config(["main", "api_url"])
+        self.group_by = self._config(["group_by"], default={})
+        self.hosts_vars = self._config(["hosts_vars"], default={})
 
         # Get value based on key.
         self.key_map = {
@@ -134,6 +98,66 @@ def __init__(self, script_args, script_config_data):
             "ip": "address"
         }
 
+    def _get_value_by_path(self, source_dict, key_path,
+                           ignore_key_error=False, default=""):
+        """Get key value from nested dict by path.
+
+        Args:
+            source_dict: The dict that we look into.
+            key_path: A list has the path of key. e.g. [parent_dict, child_dict, key_name].
+            ignore_key_error: Ignore KeyError if the key is not found in provided path.
+            default: Set default value if the key is not found in provided path.
+
+        Returns:
+            If key is found in provided path, it will be returned.
+            If ignore_key_error is True, None will be returned.
+            If default is defined and key is not found, default will be returned.
+        """
+
+        key_value = ""
+        try:
+            # Reduce key path, where it get value value from nested dict.
+            # a replacement for buildin reduce function.
+            for key in key_path:
+                if isinstance(source_dict.get(key), dict) and len(key_path) > 1:
+                    source_dict = source_dict.get(key)
+                    key_path = key_path[1:]
+                    self._get_value_by_path(source_dict, key_path,
+                                            ignore_key_error=ignore_key_error, default=default)
+                else:
+                    key_value = source_dict[key]
+
+        # How to set the key value, if the key was not found.
+        except KeyError as key_name:
+            if default:
+                key_value = default
+            elif not default and ignore_key_error:
+                key_value = None
+            elif not key_value and not ignore_key_error:
+                sys.exit("The key %s is not found. Please remember, Python is case sensitive." % key_name)
+        return key_value
+
+    def _config(self, key_path, default=""):
+        """Get value from config var.
+
+        Args:
+            key_path: A list has the path of the key.
+            default: Default value if the key is not found.
+
+        Returns:
+            The value of the key from config file or the default value.
+        """
+        config = self.script_config.setdefault("netbox", {})
+        value = self._get_value_by_path(config, key_path, ignore_key_error=True, default=default)
+
+        if value:
+            key_value = value
+
+        else:
+            sys.exit("The key '%s' is not found in config file." % ".".join(key_path))
+
+        return key_value
+
     @staticmethod
     def get_hosts_list(api_url, specific_host=None):
         """Retrieves hosts list from netbox API.
@@ -219,7 +243,7 @@ def add_host_to_inventory(self, groups_categories, inventory_dict, host_data):
                 # The groups that will be used to group hosts in the inventory.
                 for group in groups_categories[category]:
                     # Try to get group value. If the section not found in netbox, this also will print error message.
-                    group_value = get_value_by_path(data_dict, [group, key_name])
+                    group_value = self._get_value_by_path(data_dict, [group, key_name])
                     inventory_dict = self.add_host_to_group(server_name, group_value, inventory_dict)
 
         # If no groups in "group_by" section, the host will go to catch-all group.
@@ -263,7 +287,7 @@ def get_host_vars(self, host_data, host_vars):
                     # This is because "custom_fields" has more than 1 type.
                     # Values inside "custom_fields" could be a key:value or a dict.
                     if isinstance(data_dict.get(var_data), dict):
-                        var_value = get_value_by_path(data_dict, [var_data, key_name], ignore_key_error=True)
+                        var_value = self._get_value_by_path(data_dict, [var_data, key_name], ignore_key_error=True)
                     else:
                         var_value = data_dict.get(var_data)
 
@@ -346,6 +370,7 @@ def main():
     ansible_inventory = netbox.generate_inventory()
     netbox.print_inventory_json(ansible_inventory)
 
+
 # Run main.
 if __name__ == "__main__":
     main()

setup.py

@@ -5,7 +5,6 @@
 # setup.py file is part of Netbox dynamic inventory script.
 # https://github.com/AAbouZaid/netbox-as-ansible-inventory
 
-from setuptools.command.install import install as InstallCommand
 from setuptools.command.test import test as TestCommand
 from setuptools import setup, Command
 from codecs import open as openc

tests/test_netbox.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python
 from __future__ import absolute_import
 
-import os
 import sys
 import json
 import yaml
@@ -211,17 +210,6 @@ class Args(object):
 # Test Netbox utils functions.
 class TestNetboxUtils(object):
 
-    @pytest.mark.parametrize("source_dict, key_path", [
-        ({"a_key": {"b_key": {"c_key": "c_value"}}},
-         ["a_key", "b_key", "c_key"])
-    ])
-    def test_reduce_path(self, source_dict, key_path):
-        """
-        Test reduce_path function.
-        """
-        reduced_path = netbox.reduce_path(source_dict, key_path)
-        assert reduced_path == "c_value"
-
     @pytest.mark.parametrize("source_dict, key_path", [
         ({"a_key": {"b_key": {"c_key": "c_value"}}},
          ["a_key", "b_key", "c_key"])
@@ -230,7 +218,7 @@ def test_get_value_by_path_key_exists(self, source_dict, key_path):
         """
         Test get value by path with exists key.
         """
-        reduced_path = netbox.get_value_by_path(source_dict, key_path)
+        reduced_path = netbox_inventory._get_value_by_path(source_dict, key_path)
         assert reduced_path == "c_value"
 
     @pytest.mark.parametrize("source_dict, key_path", [
@@ -242,7 +230,7 @@ def test_get_value_by_path_key_not_exists(self, source_dict, key_path):
         Test get value by path with non-exists key.
         """
         with pytest.raises(SystemExit) as key_not_exists:
-            netbox.get_value_by_path(source_dict, key_path)
+            netbox_inventory._get_value_by_path(source_dict, key_path)
         assert key_not_exists
 
     @pytest.mark.parametrize("source_dict, key_path, ignore_key_error", [
@@ -254,9 +242,23 @@ def test_get_value_by_path_key_not_exists_ignore_error(self, source_dict, key_pa
         """
         Test get value by path with exists key and not ignore error.
         """
-        reduced_path = netbox.get_value_by_path(source_dict, key_path, ignore_key_error)
+        reduced_path = netbox_inventory._get_value_by_path(
+            source_dict, key_path, ignore_key_error=ignore_key_error)
         assert reduced_path is None
 
+    @pytest.mark.parametrize("source_dict, key_path, default", [
+        ({"a_key": {"b_key": {"c_key": "c_value"}}},
+         ["a_key", "b_key", "any"],
+         "default_value")
+    ])
+    def test_get_value_by_path_key_not_exists_with_default_value(self, source_dict, key_path, default):
+        """
+        Test get value by path with exists key and not ignore error.
+        """
+        reduced_path = netbox_inventory._get_value_by_path(
+            source_dict, key_path, default=default)
+        assert reduced_path is "default_value"
+
     @pytest.mark.parametrize("yaml_file", [
         "netbox.yml"
     ])