rougail/src/rougail/convert/collect.py

695 lines
28 KiB
Python

"""Takes a bunch of Rougail YAML dispatched in differents folders
as an input and outputs a Tiramisu's file.
Silique (https://www.silique.fr)
Copyright (C) 2026
This program is free software: you can redistribute it and/or modify it
under the terms of the GNU Lesser General Public License as published by the
Free Software Foundation, either version 3 of the License, or (at your
option) any later version.
This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
import logging
from warnings import warn
from pydantic import ValidationError
from copy import deepcopy
from typing import (
Any,
Iterator,
Literal,
Optional,
)
from tiramisu.error import display_list
from .object_model import (
CALCULATION_TYPES,
CALCULATION_PROPERTY_TYPES,
PROPERTY_ATTRIBUTE,
PARAM_TYPES,
AnyParam,
)
from ..i18n import _
from ..error import DictConsistencyError
class CollectFamily:
def family_collect(self) -> None:
if self.parameters is None:
self.parameters = {}
if self.user_type == "dynamic":
# FIXME code here only for support 1.0 dynamic flavor
# DictConsistencyError should be place in annotation
if "{{ identifier }}" not in self.name:
if "{{ suffix }}" in self.name:
self.name = self.name.replace("{{ suffix }}", "{{ identifier }}")
self.path = self.path.replace("{{ suffix }}", "{{ identifier }}")
elif "variable" in self.parameters:
self.name += "{{ identifier }}"
self.path += "{{ identifier }}"
elif self.raises:
msg = f'dynamic family name must have "{{{{ identifier }}}}" in his name for "{self.path}"'
raise DictConsistencyError(msg, 13, self.sources)
if self.version == "1.0":
if self.raises and "variable" not in self.parameters:
raise DictConsistencyError(
f'dynamic family must have "variable" attribute for "{self.path}"',
101,
self.sources,
)
if self.raises and "dynamic" in self.parameters:
raise DictConsistencyError(
'variable and dynamic cannot be set together in the dynamic family "{self.path}"',
100,
self.sources,
)
self.parameters["dynamic"] = {
"type": "variable",
"variable": self.parameters["variable"],
"propertyerror": False,
"allow_none": True,
}
del self.parameters["variable"]
if "dynamic" not in self.parameters and "variable" in self.parameters:
self.parameters["dynamic"] = {
"type": "variable",
"variable": self.parameters["variable"],
"propertyerror": False,
"allow_none": True,
}
del self.parameters["variable"]
if self.version != "1.0":
warning = f'"variable" attribute in dynamic family "{self.path}" is depreciated in {display_list(self.sources)}'
warn(
warning,
DeprecationWarning,
stacklevel=2,
)
self.object = self.objectspace.family_objects[0]
else:
self.object = self.objectspace.family_objects[-1]
self.split_param_family_children()
def split_param_family_children(
self,
) -> None:
"""Family declaration mix variable object and family attributes
Here we separate attribute (which is the correct self.parameters content and variable declaration
"""
new_parameters = {}
children = self.parameters
# when an attribute starts with "_", the variable name without this "_" is a variable
sub_variables = [key[1:] for key in self.parameters if key.startswith('_') and key[1:] in self.parameters]
# and known variables
if self.test_exists and self.path in self.objectspace.paths:
sub_variables.extend([p.rsplit('.', 1)[-1] for p in self.objectspace.parents[self.path]])
attributes = self.object["attrs"]
attributes_types = self.object["attributes_types"]
if self.types:
types_children = list(self.types.children)
else:
types_children = []
for key, value in list(children.items()):
if self.raises and not isinstance(key, str):
raise DictConsistencyError(
f'the key "{key}" is not in string format for family {self.path}',
103,
self.sources,
)
if key.startswith("_"):
# if key starts with _, it's an attribute
if key == "_type" and self.types:
children.pop(key)
else:
new_parameters[key[1:]] = children.pop(key)
elif key in types_children or \
key not in attributes or \
key in sub_variables or \
not self.parameter_is_an_attributes(key, value, attributes_types):
if key == "type" and self.types and key not in types_children:
children.pop(key)
else:
new_parameters[key] = children.pop(key)
if self.version != "1.0" and not new_parameters and self.comment:
new_parameters["description"] = self.comment
self.children = children
self.parameters = new_parameters
class CollectVariable:
def variable_collect(self) -> None:
obj = self.parameters
if self.version == "1.0" or isinstance(obj, dict):
if obj is None:
# only with self.version to 1.0
obj = {}
self.parameters = obj
self.parse_type_params()
self.parse_secret_manager()
if self.raises and not self.check_no_extra_keys:
extra_attrs = set(obj) - self.object["attrs"]
if extra_attrs:
raise DictConsistencyError(
f'"{self.path}" is not a valid variable, there are additional '
f'attributes: "{", ".join(extra_attrs)}"',
65,
self.sources,
)
else:
self.parameters = {"default": obj}
if self.comment:
self.parameters["description"] = self.comment
def parse_type_params(self):
"""Parse variable params"""
if "params" not in self.parameters:
return
params = self.parameters["params"]
if isinstance(params, list):
# the conversion has already be done (in types purpose)
for param in params:
if self.raises and not isinstance(param, AnyParam):
raise DictConsistencyError(
_("params must be a dict for {0}").format(self.path),
55,
self.sources,
)
return
elif self.raises and not isinstance(params, dict):
raise DictConsistencyError(
_("params must be a dict for {0}").format(self.path),
55,
self.sources,
)
new_params = []
namespace = self.objectspace.namespace
for key, val in params.items():
try:
new_params.append(
AnyParam(
key=key,
value=val,
type="any",
path=None,
attribute=None,
family_is_dynamic=None,
namespace=namespace,
xmlfiles=self.sources,
)
)
except ValidationError as err:
if self.raises:
raise DictConsistencyError(
_('"{0}" has an invalid "params" for {1}: {2}').format(
key, self.path, err
),
54,
self.sources,
) from err
self.parameters["params"] = new_params
def parse_secret_manager(self):
"""Parse variable secret_manager"""
if "secret_manager" not in self.parameters:
return
secret_manager = {
"type": "jinja",
"jinja": self.objectspace.secret_pattern,
"params": self.parameters["secret_manager"],
}
self.set_calculation(
"secret_manager",
secret_manager,
)
class CollectType:
"""Determine the option type
1/ self.option_type must be "variable" or "family" option type
needs to determinate if the option is a variable or a family
an option is a family if:
- user specified a know family type: family, sequence (user must set it explicitly) or dynamic)
- has a family custom type
- has "dynamic" attribute (a variable could not has dynamic attribut)
- it's already a family (so we are certainly in redefine purpose
otherwise it's a variable
2/ self.user_type must be a more specilized variable or family type defined by the user
"""
def set_option_types(self):
if isinstance(self.parameters, dict):
self.set_user_type()
self.set_dynamic_user_type()
else:
self.short_hand_variable()
self.set_user_redefined()
self.variable_in_sequence()
self.family_or_variable()
if self.user_type:
logging.info("family_or_variable: %s is a %s (%s)", self.path, self.option_type, self.user_type)
else:
logging.info("family_or_variable: %s is a %s", self.path, self.option_type)
def set_user_type(
self,
) -> None:
"""Check 'type' attributes and determine the option type"""
for type_name in ["_type", "type"]:
if type_name not in self.parameters or not isinstance(self.parameters[type_name], str):
continue
self.user_type = self.parameters[type_name]
if self.user_type in self.objectspace.custom_family_types:
self.set_custom_type(self.objectspace.custom_family_types[self.user_type])
if self.user_type is None:
self.object = self.objectspace.family_objects[-1]
else:
self.set_object_user_type_family()
break
elif self.user_type in self.objectspace.custom_variable_types:
self.set_custom_type(self.objectspace.custom_variable_types[self.user_type])
if self.user_type is None:
self.object = self.objectspace.variable_objects[0]
else:
self.set_object_user_type_variable()
break
self.object = None
self.set_object_user_type_family()
if not self.object:
self.set_object_user_type_variable()
if self.raises and not self.object:
msg = _('cannot determine the type of the {0} "{1}"').format(self.user_type, self.path)
raise DictConsistencyError(msg, 43, self.sources)
break
def set_custom_type(self, custom: dict) -> None:
if self.raises and self.test_exists and self.path in self.objectspace.paths:
msg = f'cannot redefine "{self.path}" object to a custom type'
raise DictConsistencyError(msg, 64, self.sources)
self.types = custom
self.types.name = self.name
self.types.path = self.path
self.types.namespace = self.namespace
self.option_type = self.types.option_type
self.user_type = self.types.user_type
self.sources = self.types.sources + self.sources
self.object = self.types.object
if self.option_type == "variable" and "type" in self.parameters:
self.parameters.pop("type")
def set_object_user_type_family(self):
for obj in self.objectspace.family_objects:
if self.user_type in obj["types"]:
self.option_type = "family"
self.object = obj
return
def set_object_user_type_variable(self):
for obj in self.objectspace.variable_objects:
if self.user_type in obj["types"]:
self.option_type = "variable"
self.object = obj
return
def set_dynamic_user_type(self) -> None:
if self.user_type:
return
for type_name in ["_dynamic", "dynamic"]:
if type_name in self.parameters and isinstance(self.parameters[type_name], (list, dict)):
self.user_type = "dynamic"
self.option_type = "family"
break
def set_user_redefined(self) -> None:
if not self.test_exists or self.path not in self.objectspace.paths:
return
# it's already a variable or a family
old_option_type = self.option_type if self.user_type else None
self.option_type = "family" if self.path in self.objectspace.families else "variable"
if self.raises and old_option_type and self.option_type != old_option_type:
msg = f'the {old_option_type} "{self.path}" is redefine as a {self.option_type}, which is not allowed'
raise DictConsistencyError(msg, 11, self.sources)
if self.user_type:
return
self.user_type = self.objectspace.paths[self.path].type
if self.option_type == "family":
for obj in self.objectspace.family_objects:
if self.user_type in obj["types"]:
self.object = obj
break
else:
self.object = self.objectspace.variable_objects[0]
else:
for obj in self.objectspace.variable_objects:
if self.user_type in obj["types"]:
self.object = obj
break
else:
self.object = self.objectspace.variable_objects[-1]
def short_hand_variable(self):
if self.option_type:
return
# it's something like: "my_variable:" or "my_variable: True"
self.option_type = "variable"
self.object = self.objectspace.variable_objects[0]
def variable_in_sequence(self):
# in a sequence there have only variables
if not self.option_type and self.family_is_sequence:
self.option_type = "variable"
if self.raises and not self.find_variable_object():
msg = f'"{self.path}" seems to be a family, which is not allowed in a family'
raise DictConsistencyError(msg, 17, self.sources)
def family_or_variable(
self,
) -> Literal["variable", "family"]:
"""check if all attributes are attributes of a variable"""
if self.option_type:
return
if not self.find_variable_object():
self.option_type = "family"
def find_variable_object(self) -> bool:
attrs = set(self.parameters)
for variable in self.objectspace.variable_objects:
if not attrs - variable["attrs"] and self.check_variable_parameters(variable):
self.option_type = "variable"
self.check_no_extra_keys = True
self.object = variable
return True
return False
def check_variable_parameters(self, variable):
attributes_types = variable["attributes_types"]
attrs = variable["attrs"]
for key, value in list(self.parameters.items()):
if key not in attrs:
return False
if not self.parameter_is_an_attributes(key, value, attributes_types):
return False
return True
def parameter_is_an_attributes(self, key: str, value: Any, attributes_types: dict) -> bool:
if value is None:
return True
if key in attributes_types["literals"] and value in attributes_types["literals"][key]:
return True
for k, typ in [("strings", str), ("booleans", bool), ("integers", int), ("floats", float)]:
if key in attributes_types[k] and isinstance(value, typ):
return True
if isinstance(value, dict):
if key in attributes_types["calculation"]:
return self.is_calculation(key, value, attributes_types=attributes_types)
if key in attributes_types["params"]:
return True
if isinstance(value, list):
current_value = value.copy()
for k, typ in [("strings", str), ("booleans", bool), ("integers", int), ("floats", float)]:
if key in attributes_types["lists"][k]:
for idx, val in reversed(list(enumerate(current_value))):
if val is not None and not isinstance(val, typ):
continue
del current_value[idx]
if not current_value:
return True
if key in attributes_types["lists"]["calculation"]:
for val in current_value:
if isinstance(val, dict) and not self.is_calculation(key, val, inside_list=True, attributes_types=attributes_types):
return False
return True
return False
class Collect(CollectType, CollectFamily, CollectVariable):
def __init__(
self,
objectspace: "ParserVariable",
name: str,
subpath: Optional[str],
parameters: dict,
comment: Optional[str],
parent_option: Optional[Collect],
*,
raises: bool=True,
test_exists: bool=True,
) -> None:
self.sources_types = None
self.types = None
self.name = name
if not subpath:
path = name
else:
path = f"{subpath}.{name}"
self.raises = raises
self.test_exists = test_exists
if self.raises and name.startswith("_"):
msg = f'the variable or family "{self.path}" is incorrect, it must not starts with "_" character'
raise DictConsistencyError(msg, 16, obj["sources"])
self.path = path
self.subpath = subpath
self.parameters = parameters
self.comment = comment
self.objectspace = objectspace
self.namespace = objectspace.namespace
self.sources = objectspace.sources.copy()
self.user_type = None
self.option_type = None
if parent_option is not None and parent_option.types is not None and self.name in parent_option.types.children:
self.set_custom_type(parent_option.types.children[self.name])
self.check_no_extra_keys = False
if self.test_exists and path in objectspace.paths:
for source in reversed(self.objectspace.paths[path].xmlfiles):
self.sources.insert(0, source)
if parent_option:
self.family_is_sequence = parent_option.user_type in ["leadership", "sequence"]
self.family_is_dynamic = parent_option.family_is_dynamic
self.parent_dynamic = parent_option.parent_dynamic
else:
self.family_is_sequence = False
self.family_is_dynamic = False
self.parent_dynamic = None
self.version = self.objectspace.version
self.set_option_types()
if self.user_type == "dynamic":
self.family_is_dynamic = True
self.parent_dynamic = subpath
if self.option_type == "family":
self.family_collect()
else:
self.variable_collect()
self.convert_calculated_parameters()
self.exists = self.is_exists()
self.redefine = self.is_redefine()
def convert_calculated_parameters(
self,
):
"""Parse option attributes and convert calculation"""
obj = self.parameters
for key, value in obj.items():
if self.is_calculation(
key,
value,
):
try:
self.set_calculation(
key,
value,
)
except ValidationError as err:
if self.raises:
raise DictConsistencyError(
_('the {0} "{1}" has an invalid "{2}": {3}').format(self.option_type, self.path, key, err),
84,
self.sources,
) from err
continue
if not isinstance(value, list):
continue
for idx, val in enumerate(value):
if not self.is_calculation(
key,
val,
inside_list=True,
):
continue
try:
self.set_calculation(
key,
val,
inside_list=True,
index=idx,
)
except ValidationError as err:
raise Exception(
f'the {option.type} "{option.path}" in "{display_list(option.sources)}" has an invalid "{key}" '
f"at index {idx}: {err}"
) from err
def is_dict(self, key: str, value: Any, *, attributes_types: Optional[dict]=None) -> bool:
"""it's a dict, so it's a new variables!"""
return isinstance(value, dict) and not self.is_calculation(key, value, attributes_types=attributes_types)
def is_calculation(
self,
attribute: str,
value: dict,
*,
attributes_types: Optional[dict]=None,
inside_list: bool=False,
):
"""Check if it's a calculation"""
if not isinstance(value, dict):
return False
if attributes_types is None:
attributes_types = self.object["attributes_types"]
if inside_list:
attributes_types = attributes_types["lists"]["calculation"]
else:
attributes_types = attributes_types["calculation"]
if attribute not in attributes_types:
return False
return self.manage_calculation_type(value)
def manage_calculation_type(self, value) -> bool:
"""Return true if the dict type is a calculation
If user set the type, it's easy, we just have to check if it's a known calculation type
Otherwise if we have an attribute with a known calculation type, we assume that is de calculation type
"""
if "type" in value:
return value["type"] in CALCULATION_TYPES
# auto set type
typ = set(CALCULATION_TYPES) & set(value)
# XXX variable could have identifier
if typ == {"variable", "identifier"}:
typ = {"variable"}
# XXX variable is also set to information
if typ == {"variable", "information"}:
typ = {"information"}
if len(typ) == 1:
value["type"] = typ.pop()
return True
return False
def set_calculation(
self,
attribute: str,
value: dict,
*,
obj: Optional[dict] = None,
inside_list: bool = False,
index: int = None,
):
"""This variable is a calculation"""
if obj is None:
obj = self.parameters
calculation_object = value.copy()
typ = calculation_object.pop("type")
namespace = self.objectspace.namespace
calculation_object["attribute_name"] = attribute
calculation_object["inside_list"] = inside_list
calculation_object["version"] = self.version
calculation_object["namespace"] = namespace
calculation_object["xmlfiles"] = self.sources
#
self.set_identifier_calculation_in_default_attribut(attribute, calculation_object, inside_list, index)
self.set_params_calculation(calculation_object, attribute)
#
return_type = calculation_object.get("return_type")
if return_type and return_type not in self.objectspace.variable_objects[0]["types"]:
raise Exception(
f'unknown "return_type" in {attribute} of variable "{self.path}"'
)
#
if self.raises and typ == "identifier" and not self.family_is_dynamic:
msg = f'identifier calculation for "{attribute}" in "{self.path}" cannot be set variable is not in dynamic family'
raise DictConsistencyError(msg, 53, self.sources)
if attribute in PROPERTY_ATTRIBUTE:
calc = CALCULATION_PROPERTY_TYPES[typ](**calculation_object)
else:
calc = CALCULATION_TYPES[typ](**calculation_object)
if index is None:
obj[attribute] = calc
else:
obj[attribute][index] = calc
def set_identifier_calculation_in_default_attribut(self, attribute: str, calculation_object: dict, inside_list: bool, index: Optional[int]) -> None:
if attribute != "default" or "identifier" not in calculation_object:
return
identifier = calculation_object["identifier"]
if self.is_calculation(
"identifier",
identifier,
attributes_types=self.objectspace.variable_objects[0]["attributes_types"],
inside_list=inside_list,
):
self.set_calculation("identifier", identifier, obj=calculation_object, inside_list=inside_list, index=index)
def set_params_calculation(self, calculation_object: dict, attribute: str) -> None:
if "params" not in calculation_object:
return
params = []
for key, value in calculation_object["params"].items():
if isinstance(value, dict) and "type" not in value:
self.set_param_type(value)
if not isinstance(value, dict) or "type" not in value:
param_typ = "any"
value = {
"value": value,
"type": "any",
}
else:
if self.version == "1.0" and value["type"] == "suffix":
value["type"] = "identifier"
param_typ = value["type"]
value["key"] = key
value["path"] = self.path
value["family_is_dynamic"] = self.family_is_dynamic
value["attribute"] = attribute
value["namespace"] = self.objectspace.namespace
value["xmlfiles"] = self.sources
if self.raises and param_typ not in PARAM_TYPES:
raise DictConsistencyError(
f'unknown type "{param_typ}" for "{self.path}"',
52,
self.sources,
)
try:
params.append(PARAM_TYPES[param_typ](**value))
except ValidationError as err:
if self.raises:
raise DictConsistencyError(
f'"{attribute}" has an invalid "{key}" for "{self.path}": {err}',
29,
self.sources,
) from err
calculation_object["params"] = params
def set_param_type(self, val):
# auto set type
param_typ = set(CALCULATION_TYPES) & set(val)
# XXX variable is also set to information
if param_typ == {"variable", "information"}:
param_typ = {"information"}
if len(param_typ) == 1:
val["type"] = list(param_typ)[0]
def is_redefine(self):
if self.path in self.objectspace.paths and self.option_type == "family" and not self.parameters:
# allow loading family with no attribute loaded
return True
if self.types:
return True
return self.parameters.pop("redefine", False) or (self.types and list(self.parameters) in [[], ["default"]])
def is_exists(self):
if self.version == "1.0" and self.option_type == "family":
return None
return self.parameters.pop("exists", None)