feature: implemented cli

.gitignore

@@ -5,7 +5,7 @@
 /.ipynb_checkpoints/
 
 # ignore python cache
-*/__pycache__/*
+**/*/__pycache__/
 *.egg-info/
 /.eggs/
 build/

setup.py

@@ -22,8 +22,8 @@
         "License :: OSI Approved :: MIT License",
         "Operating System :: OS Independent",
     ],
-    package_dir={"": "configen"},
-    packages=setuptools.find_packages(where="configen"),
+    package_dir={"": "src"},
+    packages=setuptools.find_packages(where="src"),
     python_requires=">=3.8",
     setup_requires=["setuptools_scm"],
     install_requires=required,

src/configen/__init__.py

@@ -0,0 +1,8 @@
+"""
+.. include:: ../README.md
+"""
+from importlib.metadata import version
+
+__author__ = "Ling"
+__email__ = "[email protected]"
+__version__ = version("configen")

configen/base_parser.py → src/configen/base_parser.py

@@ -2,10 +2,14 @@
 from __future__ import annotations
 
 import abc
+import logging
 import os
 import re
-from collections import deque
-from typing import Tuple, Union
+from typing import Tuple, Union, Optional, Any, Dict
+
+from configen.utils import merge
+
+logger = logging.getLogger(__name__)
 
 
 class Parser:
@@ -16,13 +20,13 @@ class Parser:
     config: dict = {}
     """The loaded config."""
 
-    def __init__(self, config: Union[dict, None] = None):
+    def __init__(self, config: Optional[dict] = None):
         """Initiate object with optional initial config."""
         if config is not None:
             assert isinstance(
                 config, dict
             ), f"Expected config to be dict get {type(config)}"
-        self.config = config
+            self.config = config
 
     def __eq__(self, parser: object) -> bool:
         """Compares if given parser is same as self."""
@@ -48,7 +52,8 @@ def _append_extension(self, input_path: str) -> str:
 
             `_check_extension("config.json")` -> config.json
         """
-        assert isinstance(input_path, str), f"expected type str got {type(input_path)}"
+        assert isinstance(input_path, str),\
+            f"expected type str got {type(input_path)}"
 
         filename, file_extension = os.path.splitext(input_path)
         if file_extension != "." + self.extension:
@@ -102,7 +107,52 @@ def _search_match(name: str, ignored: Tuple[str]) -> bool:
                 return True
         return False
 
-    def load(self, config: Union[str, dict, None], ignored: Tuple[str] = ()) -> Parser:
+    def join(
+            self,
+            curr_config: Dict[str, Any],
+            filepath: str,
+            ignored: Tuple[str]):
+        """Joins config.
+
+        Params:
+            curr_config: the existing loaded config
+            filepath: file path to the new config to be loaded
+            ignored: list of file names to be ignored
+
+        Returns:
+            updated config
+        """
+        # the current loaded filepath
+        logger.debug(f"{filepath=}")
+        # base folder will be used as the key
+        base_folder = os.path.basename(filepath)
+        filename, file_extension = os.path.splitext(filepath)
+
+        if self._search_match(filepath, ignored):
+            # ignore the file if it's in the ignored list
+            return curr_config
+
+        if file_extension == "." + self.extension:
+            # load the file if it's of the config format
+            logger.info(f"{'='*5} Reading {filepath}")
+            new_config = self._load_method(filepath)
+            curr_config = merge(curr_config, new_config)
+            logger.debug(f"New config = {curr_config}")
+
+        elif os.path.isdir(filepath):
+            # if the path is a folder, iteratively add the folder files
+            files = os.listdir(filepath)
+            for file in files:
+                new_path = os.path.join(filepath, file)
+                curr_config[base_folder] = self.join(
+                    curr_config.get(base_folder, {}),
+                    new_path, ignored)
+        return curr_config
+
+    def load(
+        self, config: Union[str, dict, None], ignored: Tuple[str] = ("",),
+        add_path: bool = False
+    ) -> Parser:
         """Loads the config (single, or multiple files, or dict).
 
         Params:
@@ -113,6 +163,7 @@ def load(self, config: Union[str, dict, None], ignored: Tuple[str] = ()) -> Pars
             3. dictionary containing the config itself
 
             ignored: list of regex match strings to ignore in file names
+            add_path: if to add the config filepath
 
         Returns:
             self with the config loaded in memory
@@ -132,62 +183,40 @@ def load(self, config: Union[str, dict, None], ignored: Tuple[str] = ()) -> Pars
         if isinstance(ignored, str):
             ignored = (ignored,)
 
-        assert isinstance(ignored, tuple), "expected ignored as tuple, got {type(ignored)}"
+        assert isinstance(
+            ignored, tuple
+        ), f"expected ignored as tuple, got {type(ignored)}"
 
         # if config is None, then remove the stored config
         if config is None:
-            self.config = None
+            self.config = {}
             return self
 
         # if given dictionary then stores it and end
-        if isinstance(config, dict):
+        elif isinstance(config, dict):
+            logger.info(f"{'='*5} Loading dictionary")
             self.config = config
             return self
 
         filename, file_extension = os.path.splitext(config)
         # if the config is a single config
         if file_extension == "." + self.extension:
+            logger.info(f"{'='*5} Loading single file")
             self.config = self._load_method(config)
             return self
 
-        # idea: iterate through the root folder, parse all configs
-        # stores the folder into a queue, then literately retrieve queue to
-        # maintain folder hierarchy
-        files = os.listdir(config)
-        queue: deque[str] = deque()
-
         if self.config is None:
             self.config = {}
 
-        # first iteration to get the depth 1 keys and folders
-        for file in files:
-            # skip those in the ignored
-            if self._search_match(file, ignored):
-                continue
-            filename, file_extension = os.path.splitext(file)
-            filepath = os.path.join(config, file)
-            if file_extension == "." + self.extension:
-                self.config.update(self._load_method(filepath))
-            elif os.path.isdir(filepath):
-                queue.append(filepath)
-
-        # while queue is not empty, repeat the procedure
-        while queue:
-            folder = queue.pop()
-            files = os.listdir(folder)
-            for file in files:
-                # skip those in the ignored
-                if self._search_match(file, ignored):
-                    continue
-                filename, file_extension = os.path.splitext(file)
-                filepath = os.path.join(folder, file)
-                base_folder = os.path.basename(folder)
-                if base_folder not in self.config:
-                    self.config[base_folder] = {}
-                if file_extension == "." + self.extension:
-                    self.config[base_folder].update(self._load_method(filepath))
-                elif os.path.isdir(filepath):
-                    queue.append(filepath)
+        self.config = self.join(self.config, config, ignored=ignored)
+
+        # in some occasions the folder containing the config will become the
+        # level1 key, fix this by loading the values instead
+        base_folder = os.path.basename(config)
+        if base_folder in self.config:
+            self.config = self.config[base_folder]
+        if add_path:
+            self.config["config_path"] = config
 
         return self
 
@@ -232,7 +261,7 @@ def write(self, filename: str, config: Union[str, dict, None] = None) -> Parser:
         return self
 
     def convert(
-        self, filename: str, parser: type[Parser], config_path: Union[str, None] = None
+        self, filename: str, parser: type[Parser], config_path: Optional[str] = None
     ) -> Parser:
         """Converts the config file into another file extension.
 

src/configen/cli.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python
+"""Entry point for program"""
+import argparse
+import logging
+import json
+import sys
+
+from configen.parsers import JsonParser, YamlParser
+from configen.utils import merge
+
+
+logger = logging.getLogger(__name__)
+
+
+def entry(args):
+    """Command line interface entry point.
+
+    Example:
+        configen config.json
+    """
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "path", help="path to the config file/folder", type=str)
+    parser.add_argument(
+        "-o", "--output", help="path to save the loaded config", type=str)
+    parser.add_argument(
+        "-f", "--format", help="config output format",
+        choices=["json", "yaml"], default="json", type=str)
+    parser.add_argument(
+        "-i", "--ignored",
+        nargs="*",
+        help="list of files to be ignored, support regex", type=str)
+    parser.add_argument(
+        "-v", "--verbose",
+        help="debug level", type=str, default="INFO")
+    parser.add_argument(
+        "-a", "--append",
+        help="append arbitrary dictionary in json format", type=str
+    )
+    parser.add_argument(
+        "-r", "--read",
+        nargs="*",
+        help="which filetype to read", type=str, default=["*"]
+    )
+    args = parser.parse_args(args)
+
+    # variables needed
+    config_path = args.path
+    output_path = args.output
+    output_format = args.format
+    ignored = tuple(args.ignored) if args.ignored else ()
+    append_dict = json.loads(args.append) if args.append else {}
+    read_format = args.read
+
+    logging.basicConfig(
+        datefmt='%m/%d/%Y %I:%M:%S %p',
+        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+        level=args.verbose
+    )
+    mega_config = {}
+
+    # initate parsers
+    config_parser_dict = {
+        "json": JsonParser(),
+        "yaml": YamlParser()
+    }
+    if read_format == "*":
+        read_format = list(config_parser_dict.keys())
+
+    # load config
+    for config_parser_name in read_format:
+        config_parser = config_parser_dict[config_parser_name]
+        config = config_parser.load(config=config_path, ignored=ignored)
+        merge(mega_config, config.config)
+
+    # override the append dict
+    if append_dict:
+        logger.info(f"Override dictionary value with {append_dict}")
+    mega_config.update(append_dict)
+
+    # save config
+    if output_path is not None:
+        config_parser_dict[output_format].write(output_path, mega_config)
+
+
+if __name__ == "__main__":
+    logger.info("Start Configen command line interface")
+    entry(sys.argv[1:])

src/configen/parsers/__init__.py

@@ -0,0 +1,6 @@
+from configen.parsers.json_parser import JsonParser
+from configen.parsers.yaml_parser import YamlParser
+
+parser_list = [JsonParser, YamlParser]
+
+__all__ = ["JsonParser", "YamlParser"]

configen/yaml_parser.py → src/configen/parsers/yaml_parser.py

@@ -1,27 +1,27 @@
 from ruamel.yaml import YAML
 from configen.base_parser import Parser
 
-_yaml = YAML()
-_yaml.indent(mapping=2, sequence=4, offset=2)
-
 
 class YamlParser(Parser):
     """Yaml parser."""
     extension = "yml"
 
+    _yaml = YAML()
+    _yaml.indent(mapping=2, sequence=4, offset=2)
+
     def _write_method(self, filename: str) -> Parser:
         # check if the given path ends with a yaml file extension
         filename = self._append_extension(filename)
 
         with open(filename, "w") as file:
-            _yaml.dump(self.config, file)
+            self._yaml.dump(self.config, file)
 
         return self
 
     def _load_method(self, filename: str) -> dict:
         filename = self._append_extension(filename)
 
         with open(filename, "r") as file:
-            config = _yaml.load(file.read())
+            config = self._yaml.load(file.read())
 
         return config