Merge remote-tracking branch 'origin/main'

# Conflicts:
#	confhub/builder.py
#	confhub/core/fields.py

confhub/__init__.py

@@ -1,5 +1,6 @@
-from confhub.core.fields import field
+from confhub.core.fields import field, exclude
 from confhub.core.block import BlockCore
 from confhub.reader import Confhub
 
-__all__ = ["field", "BlockCore", "Confhub"]
+
+__all__ = ["field", "exclude", "BlockCore", "Confhub"]

confhub/builder.py

@@ -1,70 +1,122 @@
 from pathlib import Path
-from typing import List, Any, Dict
+from typing import List, Any, Dict, Type
 
 import yaml
 import structlog
 
 from confhub.core.block import BlockCore
+from confhub.core.error import ConfhubError
 from confhub.core.fields import ConfigurationField
 from confhub.utils.gitignore import add_to_gitignore
 
 logger: structlog.BoundLogger = structlog.get_logger("confhub")
 
 
+def has_configuration_fields(select_class: BlockCore) -> bool:
+    if any(isinstance(item, ConfigurationField) for item in select_class.__dict__.values()):
+        return True
+    if any(isinstance(item, ConfigurationField) for item in select_class.__class__.__dict__.values()):
+        return False
+    raise ConfhubError("Cannot find field in model object", select_class=select_class)
+
+
 class ConfigurationBuilder:
     def __init__(self, *blocks: BlockCore):
-        self.blocks = [*blocks]
-        self.datafiles: Dict[str, Any] = self.generate_filenames()
-
-    def generate_filenames(self) -> Dict[str, Any]:
-        _datafiles = {'settings': {}, '.secrets': {}}
-
-        def add_field_to_datafiles(field_name: str, field: ConfigurationField, parent_path: List[str]) -> None:
-            if field.secret:
-                target = _datafiles['.secrets']
-            elif field.filename:
-                if field.filename not in _datafiles:
-                    _datafiles[field.filename] = {}
-                target = _datafiles[field.filename]
-            else:
-                target = _datafiles['settings']
-
-            current = target
-            for part in parent_path:
-                if part not in current:
-                    current[part] = {}
-                current = current[part]
-
-            if not isinstance(field.data_type, BlockCore):
-                current[field_name] = field.get_default_value()
-
-        def process_block(_block: BlockCore, parent_path: List[str]) -> None:
-            current_path = parent_path + [_block.__block__]
-
-            for field_name, field in _block.__dict__.items():
-                if isinstance(field, ConfigurationField):
-                    add_field_to_datafiles(field_name, field, current_path)
-                elif isinstance(field, BlockCore):
-                    process_block(field, current_path)
-
+        self.blocks = list(blocks)
+        self.datafiles: Dict[str, Any] = {'settings': {}, '.secrets': {}}
+        self.generate_filenames()
+
+    def data_typing(self, field: ConfigurationField) -> Any:
+        if isinstance(field.data_type, BlockCore):
+            nested_block = {
+                nested_field_name: self.data_typing(nested_field)
+                for nested_field_name, nested_field in field.data_type.__class__.__dict__.items()
+                if isinstance(nested_field, ConfigurationField)
+            }
+            return [nested_block] if field.is_list else nested_block
+        return [field.get_default_value()] if field.is_list else field.get_default_value()
+
+    def new_nested(self, nested_model: Type[BlockCore]):
+        nested_block = {nested_model.__block__: {}}
+        for nested_field_name, nested_field in nested_model.__dict__.items():
+            if isinstance(nested_field, ConfigurationField):
+                nested_block[nested_model.__block__][nested_field_name] = (
+                    [self.new_nested(nested_field.data_type.__class__)] if isinstance(nested_field.data_type, BlockCore) else
+                    self.data_typing(nested_field)
+                )
+            elif isinstance(nested_field, BlockCore):
+                nested_block[nested_model.__block__][nested_field_name] = [self.new_nested(nested_field.__class__)]
+        return nested_block
+
+    def add_field_to_datafiles(
+            self, field_name: str, field: ConfigurationField, parent_path: List[str]
+    ) -> None:
+        if field.secret:
+            target = self.datafiles['.secrets']
+        elif field.filename:
+            target = self.datafiles.setdefault(field.filename, {})
+        else:
+            target = self.datafiles['settings']
+
+        current = target
+        for part in parent_path:
+            current = current.setdefault(part, {})
+
+        if field.is_list and isinstance(field.data_type, BlockCore):
+            current[field_name] = [self.new_nested(field.data_type.__class__)]
+        elif field.is_list:
+            current[field_name] = [field.get_default_value()]
+        elif isinstance(field.data_type, BlockCore):
+            current[field_name] = {
+                nested_field_name: self.data_typing(nested_field)
+                for nested_field_name, nested_field in field.data_type.__class__.__dict__.items()
+                if isinstance(nested_field, ConfigurationField)
+            }
+        else:
+            current[field_name] = field.get_default_value()
+
+    def process_block(self, _block: BlockCore, parent_path: List[str], parent: str = None) -> None:
+        if hasattr(_block, '__exclude__') and _block.__exclude__ and not parent:
+            return
+
+        current_path = parent_path + ([_block.__block__] if not parent else [parent])
+
+        for field_name, field in (_block.__dict__.items() if has_configuration_fields(_block) else _block.__class__.__dict__.items()):
+            if isinstance(field, ConfigurationField):
+                if isinstance(field.data_type, BlockCore):
+                    self.add_field_to_datafiles(field_name, field, current_path)
+                    self.process_block(field.data_type, current_path + [field_name] if field.is_list else current_path, parent=field_name if not field.is_list else None)
+                else:
+                    self.add_field_to_datafiles(field_name, field, current_path)
+            elif isinstance(field, BlockCore):
+                self.process_block(field, current_path, parent=field_name)
+
+    def generate_filenames(self):
         for block in self.blocks:
-            process_block(block, [])
+            if block != BlockCore:
+                self.process_block(block, [])
 
-        return _datafiles
+    @staticmethod
+    def remove_empty_dicts(data):
+        if isinstance(data, dict):
+            return {k: ConfigurationBuilder.remove_empty_dicts(v) for k, v in data.items() if v and ConfigurationBuilder.remove_empty_dicts(v)}
+        elif isinstance(data, list):
+            return [ConfigurationBuilder.remove_empty_dicts(v) for v in data if v and ConfigurationBuilder.remove_empty_dicts(v)]
+        return data
 
     def create_files(self, config_path: Path) -> None:
-        for filename, data in self.datafiles.items():
-            file_path = config_path / Path(f'{filename}.yml')
+        datafiles = self.remove_empty_dicts(self.datafiles)
+        for filename, data in datafiles.items():
+            file_path = config_path / f'{filename}.yml'
 
             if file_path.exists():
                 with open(file_path, 'r', encoding='utf-8') as file:
                     yaml_data = yaml.safe_load(file)
-
-                    for key in data:
-                        if key in yaml_data:
-                            for inner_key in data[key]:
-                                if inner_key in yaml_data[key]:
-                                    data[key][inner_key] = yaml_data[key][inner_key]
+                    if yaml_data:
+                        for key, value in data.items():
+                            if key in yaml_data and isinstance(yaml_data[key], dict):
+                                yaml_data[key].update(value)
+                                data[key] = yaml_data[key]
 
             with open(file_path, 'w', encoding='utf-8') as file:
                 yaml.dump(data, file, default_flow_style=False)

confhub/core/block.py

@@ -1,3 +1,5 @@
+from typing import Type
+
 from confhub.core.fields import ConfigurationField
 from confhub.core.parsing import parsing_value
 

confhub/core/fields.py

@@ -1,18 +1,26 @@
-from typing import Type, Any
+from typing import Any
 
 from confhub.core.types import DataTypeMapping
 
 
 class ConfigurationField:
     def __init__(
             self,
-            data_type: Type,
+            data_type: Any,
             secret: bool,
-            filename: str
+            filename: str,
+            is_list: bool,
     ) -> None:
         self.data_type = data_type
         self.secret = secret
         self.filename = filename
+        self.is_list = is_list
+
+    def __str__(self) -> str:
+        return f"ConfigurationField: [{self.data_type}]"
+
+    def __repr__(self) -> str:
+        return f"ConfigurationField: [{self.data_type}; secret={self.secret}; filename={self.filename}; is_list={self.is_list}]"
 
     def get_default_value(self) -> str:
         return f"{self.data_type.__name__}; {DataTypeMapping.get_default_value(self.data_type.__name__)}"
@@ -21,10 +29,17 @@ def get_default_value(self) -> str:
 def field(
         data_type: Any,
         secret: bool = False,
-        filename: str = None
+        filename: str = None,
+        is_list: bool = False
 ) -> ConfigurationField:
     return ConfigurationField(
         data_type=data_type,
         secret=secret,
-        filename=filename
+        filename=filename,
+        is_list=is_list
     )
+
+
+def exclude(cls):
+    cls.__exclude__ = True
+    return cls

confhub/core/parsing.py

@@ -1,3 +1,4 @@
+import sys
 from typing import Union, Any, Dict, List
 
 import yaml

confhub/core/types.py

@@ -5,7 +5,7 @@
     "str": (str, "VALUE"),
     "int": (int, "1234"),
     "float": (float, "1234.101"),
-    "bool": (bool, "true"),
+    "bool": (bool, "true")
 }
 
 

confhub/reader.py

@@ -61,4 +61,9 @@ def __load(self, *models: BlockCore, files: List[str | Path]) -> Type[dataclasse
 
             __fields_from_dataclass.append((block.__block__, type(block), dataclasses.field(default=value)))
 
-        return dataclasses.make_dataclass('Data', __fields_from_dataclass)
+        return dataclasses.make_dataclass('Data', __fields_from_dataclass)
+
+
+if __name__ == '__main__':
+    data = Confhub().models
+    print()