From c97c171f41f451e1302fef5b03701a28d1fa5930 Mon Sep 17 00:00:00 2001 From: mrjk Date: Wed, 26 Jan 2022 15:43:16 -0500 Subject: [PATCH] Lint: utils.py --- kheops/utils.py | 78 ++++++++++++++++++++++++------------------------- 1 file changed, 38 insertions(+), 40 deletions(-) diff --git a/kheops/utils.py b/kheops/utils.py index 0417e6a..dbca65e 100644 --- a/kheops/utils.py +++ b/kheops/utils.py @@ -1,59 +1,55 @@ -from pathlib import Path -from jinja2 import Template -import yaml -import json -import glob +"""Utils class""" -from jsonschema import validate, Draft7Validator, validators, exceptions import collections - - import logging +from pathlib import Path + +from jinja2 import Template +from jsonschema import Draft7Validator, validators + log = logging.getLogger(__name__) # Utils Methods # ===================== + def glob_files(path, pattern): - log.debug(f"Search glob '{pattern}' in '{path}'") - p = Path(path) - ret = p.glob(pattern) + """Return a list of path that match a glob""" + log.debug("Search glob '%s' in '%s'", pattern, path) + path = Path(path) + ret = path.glob(pattern) return [str(i) for i in ret] -def path_assemble_hier(path, sep='/'): - """Append the previous - - """ +def path_assemble_hier(path, sep="/"): + """Append the previous""" if isinstance(path, str): list_data = path.split(sep) elif isinstance(path, list): list_data = [] else: - raise Exception (f"This function only accepts string or lists, got: {path}") + raise Exception(f"This function only accepts string or lists, got: {path}") - - assert isinstance(list_data, list), f'Got: {list_data}' - ret = [] - for index, part in enumerate(list_data): - try: - prefix = ret[index - 1] - except IndexError: - prefix = f"{sep}" - prefix = "" - item = f"{prefix}{part}{sep}" - ret.append(item) + assert isinstance(list_data, list), f"Got: {list_data}" + ret = [] + for index, part in enumerate(list_data): + try: + prefix = ret[index - 1] + except IndexError: + prefix = f"{sep}" + prefix = "" + item = f"{prefix}{part}{sep}" + ret.append(item) return ret - def render_template(path, params): """Render template for a given string""" assert isinstance(params, dict), f"Got: {params}" - t = Template(path) - return t.render(**params) + tpl = Template(path) + return tpl.render(**params) # Schema Methods @@ -65,9 +61,9 @@ def _extend_with_default(validator_class): def set_defaults(validator, properties, instance, schema): - for property, subschema in properties.items(): + for prop, subschema in properties.items(): if "default" in subschema: - instance.setdefault(property, subschema["default"]) + instance.setdefault(prop, subschema["default"]) try: for error in validate_properties( @@ -77,8 +73,8 @@ def _extend_with_default(validator_class): schema, ): continue - except Exception as e: - print("CATCHED2222 ", e) + except Exception as err: + print("CATCHED2222 ", err) return validators.extend( validator_class, @@ -87,23 +83,25 @@ def _extend_with_default(validator_class): def schema_validate(config, schema): + """Validate a config against a jsonschema""" # Validate the schema DefaultValidatingDraft7Validator = _extend_with_default(Draft7Validator) try: DefaultValidatingDraft7Validator(schema).validate(config) - except Exception as e: - print(e) - p = list(collections.deque(e.schema_path)) - p = "/".join([str(i) for i in p]) - p = f"schema/{p}" + except Exception as err: + print(err) + path = list(collections.deque(err.schema_path)) + path = "/".join([str(i) for i in path]) + path = f"schema/{path}" raise Exception( - f"Failed validating {p} for resource with content: {config} with !!!!!! schema: {schema}" + f"Failed validating {path} for resource with content: {config}" ) return config def str_ellipsis(txt, length=120): + """Truncate with ellipsis too wide texts""" txt = str(txt) ret = [] for string in txt.splitlines():