Load initializers from multiple files / directory

[u1735067]

Desired Behavior

Initializers allows to load one YAML file for each model to provision, but some model can have a lot of data to load and you might want to break them in multiple files to be able to work on them. The desired behavior is to be able to load multiple files for each initializers.

Contrast to Current Behavior

Data are loaded from multiple YAML files, instead of one.

Changes Required

Adjust load_yaml.py and/or the initializer scripts to load multiple files.

Discussion: Benefits and Drawbacks

Better construction of initial data, should be backward compatible.

Proposal

I haven't tested this a lot (and the code might be a bit ugly), but here's a proposal to change a minimum amount of files (one) while implementing this idea. The load_yaml function is modified to try loading data from the requested file (model.yml) + a directory corresponding to the current model (model.d) containing .yml files with definitions (device_types.d/my_device_type1.yml for example). All files must contain the same data type (list or dict) so they can be merged and passed to the initializer script.

Modified load_yaml.py

from ruamel.yaml import YAML
from pathlib import Path

# Modified version of load_yaml.py to handle (object_type).d


def load_yaml(yaml_path: str):
	yf = Path(yaml_path)
	return merge_yaml((yf, load_yaml_recursive(yf)), (yf.with_suffix('.d'), load_yaml_recursive(yf.with_suffix('.d'))))


def load_yaml_recursive(yaml_path: Path):
	if yaml_path.is_file():
		with yaml_path.open("r") as stream:
			yaml = YAML(typ="safe")
			return yaml.load(stream)
	if yaml_path.is_dir():
		return merge_yaml(
			*((path, load_yaml_recursive(path)) for path in yaml_path.glob('*.yml'))
		)


def merge_yaml(*yamls):
	previous_entry = None
	# Check data consistency
	for path, yaml_data in yamls:
		if yaml_data is None:
			print('yaml_load: got empty data for {}'.format(path))
			continue
		current_entry = (path, _get_type(yaml_data))
		if current_entry[1] is None:
			raise Exception('yaml_load: unknown data type in {}'.format(path))
		if previous_entry is None:
			previous_entry = current_entry
		else:
			if current_entry[1] != previous_entry[1]:
				raise Exception('Content type of {} ({}) is not the same as {} ({})'.format(*current_entry, *previous_entry))
	# Merge
	yaml_datas = filter(None, map(lambda x: x[1], yamls))
	if previous_entry is not None:
		if previous_entry[1] is dict:
			return {k: v for yaml_data in yaml_datas for k, v in yaml_data.items()}
		elif previous_entry[1] is list:
			return [e for yaml_data in yaml_datas for e in yaml_data]


def _get_type(data):
	if isinstance(data, list):
		return list
	elif isinstance(data, dict):
		return dict
	else:
		return None

[cimnine]

Thanks for this feature request, and thanks for taking the time to fill in the template.

... should be backward compatible.

I'd say that any solution must be backwards compatible. I do like the current way with just a bunch of files. This serves the basic purpose of these scripts: Quickly bootstrap a working Netbox for development and demonstration purposes.

Now, I don't understand what else people are doing with this scripts, but I would be happy to consider a PR that implements your suggestion with model.d folders.

When you submit a PR, please consider these two requests of mine:

1. Enhance the test.sh file in a way such that it creates a model.d folder for any supported model, copies the default template into it, and tries to launch netbox. So that we immediately notice when something breaks.
2. Optimize the code for readability. Most of us don't deal with Python daily, so it's even more important that the code is readable. Maybe I'm already tired, but [e for yaml_data in yaml_datas for e in yaml_data] is too concise IMO.

I believe it'd be easier to understand the code if we'd first load all the YAML files of a specific model and collect it's parsed content in a list. And only then we flatten the list of lists (or list of dicts).

So based on your code I quickly hacked the following suggestion, which is not tested:

import itertools
from ruamel.yaml import YAML
from pathlib import Path

def load_yaml(yaml_base_path: str):
  yaml_path = Path(yaml_base_path)
  content_list = []
  content_list.append(load_file(yaml_path.with_suffix('.yml')))
  load_dir(content_list, yaml_path.with_suffix('.d'))
  return flatten(content_list)

def load_file(yaml_file_path: Path):
  if not yaml_file_path.is_file():
    return []
  with yaml_file_path.open("r") as stream:
    yaml = YAML(typ="safe")
    return yaml.load(stream)

def load_dir(content_list: list, yaml_dir_path: Path):
  if not yaml_dir_path.is_dir():
    return
  for yaml_file_path in yaml_dir_path.glob('*.yml')
    content_list.append(load_file(yaml_file_path))

def flatten(content_list: list):
  if len(content_list) == 0:
    return content_list
  if len(content_list) == 1:
    return content_list[0]
  if isinstance(content[0], list):
    return flatten_list(content_list)
  if isinstance(content_list[0], dict):
    return flatten_dict(content_list)
    
def flatten_list(content_list: list):
  # https://stackoverflow.com/a/952952
  flat_list = []
  for sublist in content_list:
    if not sublist:
      continue
    for item in sublist:
      flat_list.append(item)
  return flat_list

def flatten_dict(content_list: list):
  # https://stackoverflow.com/a/3495395
  flat_dict = {}
  for subdict in content_list:
    if not subdict:
      continue
    flat_dict.update(subdict)
  return flat_dict

The same as Gist: https://gist.github.com/cimnine/95f594bae302682cd12e1b38e0ba4484

Feel free to iterate on either version, yours or mine, and I'd love to see a PR for this!

[tobiasge]

Initializers were extracted to a plugin