refactor: enhance source data handling

examples/pydantic_source/default.json

@@ -1,14 +1,12 @@
 {
-    "default": {
-        "apis": {
-            "services": {
-                "value": "services inside the local file",
-                "description": "Description"
-            }
-        },
-        "common": {
-            "value": "common in local file",
-            "description": "Common Description"
+    "apis": {
+        "services": {
+            "value": "services inside the local file",
+            "description": "Description"
         }
+    },
+    "common": {
+        "value": "common in local file",
+        "description": "Common Description"
     }
 }

examples/pydantic_source/main.py

@@ -79,6 +79,7 @@ def settings_customise_sources(
                 key_prefix=KEY_PREFIX,
                 tree_name=TREE_NAME,
                 local_file="",
+                # api_with_project=False,
             ),
             env_settings,
             dotenv_settings,
@@ -134,6 +135,7 @@ def settings_customise_sources(
                 key_prefix=KEY_PREFIX,
                 tree_name=TREE_NAME,
                 local_file="",
+                # api_with_project=False,
             ),
             env_settings,
             dotenv_settings,
@@ -229,7 +231,7 @@ def settings_customise_sources(
 config_tree_with_file = RRTreeSourceLocal()
 
 
-# @app.get("/configtrees")
+@app.get("/configtrees")
 async def get_full_configtree():
     """Retrieve the full configuration tree"""
     return config_tree.model_dump()
@@ -249,4 +251,4 @@ def get_configtrees_local():
 if __name__ == "__main__":
     import uvicorn
 
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)

pydantic_source/source.py

@@ -1,10 +1,8 @@
-import json
 import ast
-from typing import Any, Type, Dict
+from typing import Any, Type, Dict, Iterable
 from benedict import benedict
-from munch import Munch
+from pathlib import Path
 
-import yaml
 from pydantic_settings import BaseSettings, PydanticBaseSettingsSource
 from pydantic.fields import FieldInfo
 
@@ -19,19 +17,18 @@ def __init__(
         config: Configuration,
         tree_name: str = "default",
         key_prefix: str = "",
+        api_with_project: bool = True,
         local_file: str = None,
     ):
         super().__init__(settings_cls)
-        self.client = Client(config=config)
-        self.tree_name = tree_name
-        self.local_file = local_file
+        self._client = Client(config=config)
+        self._tree_name = tree_name
+        self._local_file = local_file
         self._top_prefix = key_prefix
-
-        # ? Placeholder for the configuration tree data
-        self.config_tree = None
+        self._api_with_project = api_with_project
 
         # * Load the configuration tree
-        self.load_config_tree()
+        self.config_tree = self.load_config_tree()
 
         processed_data = self._process_config_tree(raw_data=self.config_tree)
 
@@ -45,60 +42,46 @@ def fetch_from_api(self):
         """
         Load the configuration tree from an external API.
         """
-        try:
-            response = self.client.get_configtree(
-                name=self.tree_name,
-                include_data=True,
-                content_types="kv",
-                with_project=False,
-            )
-            config_tree_response = Munch.toDict(response).get("keys")
-            self.config_tree = self._extract_data_api(input_data=config_tree_response)
-        except Exception as e:
-            raise ValueError(f"Failed to fetch configuration tree from API: {e}")
+        response = self._client.get_configtree(
+            name=self._tree_name,
+            include_data=True,
+            content_types="kv",
+            key_prefixes=[self._top_prefix],
+            with_project=self._api_with_project,
+        )
+        return self._extract_data_api(input_data=response["keys"].toDict())
 
     def load_from_local_file(self):
         """
         Load the configuration tree from a local JSON or YAML file.
         """
-        if not self.local_file:
-            raise ValueError("No local file path provided for configuration tree.")
+        data = {}
+        file_prefix = Path(self._local_file).stem
+        file_suffix = Path(self._local_file).suffix[1:]
 
-        try:
-            with open(self.local_file, "r") as file:
-                if self.local_file.endswith(".json"):
-                    self.config_tree = self._extract_data_local(json.load(file))
-                    self.config_tree = benedict(self.config_tree).flatten(separator="/")
-                elif self.local_file.endswith(".yaml") or self.local_file.endswith(
-                    ".yml"
-                ):
-                    self.config_tree = self._extract_data_local(yaml.safe_load(file))
-                    self.config_tree = benedict(self.config_tree).flatten(separator="/")
-                else:
-                    raise ValueError(
-                        "Unsupported file format. Use .json or .yaml/.yml."
-                    )
-        except FileNotFoundError:
-            raise ValueError(f"Local file '{self.local_file}' not found.")
-        except Exception as e:
-            raise ValueError(f"Failed to load configuration tree from file: {e}")
+        if file_suffix not in ["json", "yaml", "yml"]:
+            raise ValueError("Unsupported file format. Use .json or .yaml/.yml.")
+
+        data[file_prefix] = benedict(self._local_file, format=file_suffix)
+        data = self._split_metadata(data)
+        return benedict(data).flatten(separator="/")
 
     def load_config_tree(self):
-        if self.local_file:
-            self.load_from_local_file()
+        if self._local_file:
+            return self.load_from_local_file()
 
         else:
-            self.fetch_from_api()
+            return self.fetch_from_api()
 
     # * Methods to process the tree
-    def _extract_data_api(self, input_data: Dict[str, Any] = None) -> Dict[str, Any]:
+    def _extract_data_api(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
         return {
-            key: self._decode_and_convert(value.get("data"))
+            key: self._decode_value(value.get("data"))
             for key, value in input_data.items()
             if "data" in value
         }
 
-    def _decode_and_convert(self, encoded_data: str) -> Any:
+    def _decode_value(self, encoded_data: str) -> Any:
         decoded_data = base64.b64decode(encoded_data).decode("utf-8")
 
         try:
@@ -107,14 +90,27 @@ def _decode_and_convert(self, encoded_data: str) -> Any:
         except (ValueError, SyntaxError):
             return decoded_data
 
-    def _extract_data_local(self, input_data: Dict[str, Any] = None) -> Dict[str, Any]:
-        for key, value in input_data.items():
-            if isinstance(value, dict):
-                if "value" in value:
-                    input_data[key] = value.get("value")
-                else:
-                    self._extract_data_local(value)
-        return input_data
+    def _split_metadata(self, data: Iterable) -> Iterable:
+        """Helper function to split data and metadata from the input data."""
+        if not isinstance(data, dict):
+            return data
+
+        content = {}
+
+        for key, value in data.items():
+            if not isinstance(value, dict):
+                content[key] = value
+                continue
+
+            potential_content = value.get("value")
+
+            if len(value) == 2 and potential_content is not None:
+                content[key] = potential_content
+                continue
+
+            content[key] = self._split_metadata(value)
+
+        return content
 
     # * This method is extracting the data from the raw data and removing the top level prefix
     def _process_config_tree(self, raw_data: Dict[str, Any]) -> Dict[str, Any]: