pro-barbu-foss/kheops
pro-barbu-foss/kheops
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Lint: some parts of the code

Browse Source
This commit is contained in:
mrjk 2022-05-04 19:39:21 -04:00
parent 2707e3ae76
commit ffdbfc442b
5 changed files with 128 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
kheops/app.py
View File

@ -45,7 +45,6 @@ CONF_SCHEMA = {
"default": [],
# "arrayItem": { "$ref": "#/$defs/rules_items" },
},
"config": {
"type": "object",
"default": {},
@ -109,6 +108,7 @@ class KheopsNamespace(GenericInstance, QueryProcessor):
Kheops Namespace Class
"""
def __init__(self, app, name, config=None):
"""
Kheops Namespace Instance
@ -130,11 +130,11 @@ class KheopsNamespace(GenericInstance, QueryProcessor):
self.cache = app.cache
# Init config (from cache)
config_hash = 'conf_ns_' + dict_hash(config)
config_hash = "conf_ns_" + dict_hash(config)
try:
config = self.cache[config_hash]
log.debug("Loading namespace '%s' configuration from cache", self.name)
except KeyError as err:
except KeyError:
config = schema_validate(config, CONF_SCHEMA)
self.cache.set(config_hash, config, expire=CACHE_CONFIG_EXPIRE)
super().__init__(config)
@ -186,7 +186,6 @@ class Kheops(GenericInstance):
# self.raw_config = self.parse_conf(config)
# self.cache.set(needle, config, expire=CACHE_CONFIG_EXPIRE)
def parse_conf(self, config="kheops.yml"):
"""
Parse Kheops configuration
@ -209,12 +208,13 @@ class Kheops(GenericInstance):
elif isinstance(config, dict):
dict_conf = config
source = "dict"
self.run["conf_source"] = source
return dict_conf
def lookup(
self,
keys=None,
policy=None,
scope=None,
trace=False,
explain=False,
@ -240,7 +240,9 @@ class Kheops(GenericInstance):
for key_def in keys:
key_def = key_def or ""
assert isinstance(key_def, str), f"Expected string as key, got {type(key_def)}: {key_def}"
assert isinstance(
key_def, str
), f"Expected string as key, got {type(key_def)}: {key_def}"
# Identify namespace and key
parts = key_def.split("/")
@ -267,7 +269,7 @@ class Kheops(GenericInstance):
# Prepare output
_key = key_name
if namespace_prefix == True:
if namespace_prefix is True:
_key = key_def
ret[_key] = result
@ -285,12 +287,6 @@ class Kheops(GenericInstance):
return ret
# To clean/implement
# def DEPRECATED_dump_schema(self):

20
kheops/controllers.py
View File

@ -24,6 +24,7 @@ CACHE_QUERY_EXPIRE = 10
# Helper classes
# ------------------------
class LoadPlugin:
"""Kheops plugins loader
@ -60,7 +61,8 @@ class LoadPlugin:
# Return plugin Classe
return plugin_cls.Plugin
class BackendCandidate():
class BackendCandidate:
"""Backend Candidate
This object represents a backend candidate. It holds the value of the
@ -97,6 +99,7 @@ class Query:
# Query Processor class
# ------------------------
class QueryProcessor:
"""QueryProcessor
@ -134,9 +137,7 @@ class QueryProcessor:
# ------------------------
def query(self, key=None, scope=None, explain=False):
"""Query key with scope
"""
"""Query key with scope"""
# Look into cache
query_hash = dict_hash([self.name, key, scope])
@ -195,7 +196,6 @@ class QueryProcessor:
self.cache.set(query_hash, result, expire=CACHE_QUERY_EXPIRE)
return result
# Query parts methods
# ------------------------
@ -218,7 +218,6 @@ class QueryProcessor:
return rule
def _exec_assemble_lookups(self, lookups, query):
assert isinstance(lookups, list)
@ -258,7 +257,9 @@ class QueryProcessor:
plugin_name = plugin_def.get("module", None)
if plugin_name:
plugin = self.plugin_loader.load("scope", plugin_name)(namespace=self)
plugin = self.plugin_loader.load("scope", plugin_name)(
namespace=self
)
ret = plugin.process_items(ret, plugin_def)
new_lookups2.extend(ret)
@ -274,11 +275,12 @@ class QueryProcessor:
lookup["path"] = new_path
new_lookups3.append(lookup)
else:
log.warning("Ignore lookup item because of missing scope vars: '%s'", path)
log.warning(
"Ignore lookup item because of missing scope vars: '%s'", path
)
return new_lookups3
def _exec_backend_plugins(self, lookups, selector="matched"):
selector = "matched"
assert selector in ["last", "first", "all", "matched"]

14
kheops/plugin/backend/file.py
View File

@ -2,6 +2,7 @@
import os
import logging
# from pprint import pprint
import anyconfig
@ -11,6 +12,7 @@ from kheops.plugin.common import BackendPlugin, BackendCandidate
log = logging.getLogger(__name__)
CACHE_FILE_EXPIRE = 5
class Plugin(BackendPlugin):
"""File Backend Plugin
@ -41,13 +43,12 @@ class Plugin(BackendPlugin):
to find all of the specified format. It is better to keep this list as small
as possible.
""",
"type": "object",
"default": extensions,
"additionalProperties": {
"title": "Name of the extension with parser",
"type": "string"
}
"type": "string",
},
},
"path_prefix": {
"title": "Prefix string to append to final path",
@ -55,7 +56,7 @@ class Plugin(BackendPlugin):
String to be added at the end of the resolved path. This is useful to change
the place of the root hierarchy.
""",
"type": "string"
"type": "string",
},
"path_suffix": {
"title": "Suffix string to prepend to final path",
@ -65,12 +66,11 @@ class Plugin(BackendPlugin):
"type": "string",
"examples": [
{"path_suffix": "/ansible"},
]
],
},
},
}
}
}
_schema_props_new = {
"path": {

23
kheops/plugin/common.py
View File

@ -22,13 +22,16 @@ NoneType = type(None)
# Generic Plugin classes
# -------------------------
class KheopsPlugin:
plugin_name = None
plugin_type = None
plugin_kind = None
def __init__(self):
assert isinstance(self.plugin_name, str), f"Missing name attribute in plugin: {self.__class__}"
assert isinstance(
self.plugin_name, str
), f"Missing name attribute in plugin: {self.__class__}"
assert isinstance(self.plugin_kind, str)
config_key = f"{self.plugin_kind}_{self.plugin_name}"
@ -61,6 +64,7 @@ class KheopsItemPlugin(KheopsPlugin):
# Plugin classes
# -------------------------
class BackendPlugin(KheopsItemPlugin):
plugin_kind = "backend"
@ -140,11 +144,14 @@ class ScopePlugin(KheopsListPlugin):
self.ns = namespace
super().__init__()
# Helper classes
# -------------------------
class BackendCandidate():
class BackendCandidate:
"""Represent a backend candidate"""
def __init__(self, path=None, data=None, run=None, status=None):
assert isinstance(run, dict)
self.path = path
@ -156,7 +163,6 @@ class BackendCandidate():
return f"Status: {self.status}, Path: {self.path} => {self.data}"
class ScopeExtLoop:
"""This Scope Extension allow to loop over a lookup"""
@ -179,10 +185,15 @@ class ScopeExtLoop:
}
def loop_over(
self, lookups, conf, module_name, var_name="item", callback_context=None, callback=None
self,
lookups,
conf,
module_name,
var_name="item",
callback_context=None,
callback=None,
):
var_name = conf.get("var", var_name)
var_data_ref = conf.get("data", None)
@ -235,8 +246,6 @@ class ScopeExtLoop:
return ret
# To clean/implement

10
kheops/utils.py
View File

@ -77,6 +77,7 @@ class Default(dict):
def __missing__(self, key):
return ""
# Source: https://www.doc.ic.ac.uk/~nuric/coding/how-to-hash-a-dictionary-in-python.html
def dict_hash(dictionary: Dict[str, Any]) -> str:
"""MD5 hash of a dictionary."""
@ -104,6 +105,7 @@ def render_template_python(text, params, ignore_missing=True):
# Schema Methods
# =====================
def _extend_with_default(validator_class):
validate_properties = validator_class.VALIDATORS["properties"]
@ -113,12 +115,16 @@ def _extend_with_default(validator_class):
instance.setdefault(property, subschema["default"])
for error in validate_properties(
validator, properties, instance, schema,
validator,
properties,
instance,
schema,
):
yield error
return validators.extend(
validator_class, {"properties" : set_defaults},
validator_class,
{"properties": set_defaults},
)