Allows to register custom formats

README.md

@@ -136,6 +136,29 @@ The list of _optional_ file formats:
 In order to install confight with _optional_ formats see
 [installation](#installation) with [optional features][].
 
+## Custom Formats
+
+Other formats can be registered by name providing a function with the loader signature
+`loader(path)`.
+
+```python
+from confight import register_loader, register_extension
+
+def load_assign(path):
+    """Parse files with KEY=VALUE lines """
+    with open(path, 'r') as stream:
+        return dict(line.split('=', 1) for line in stream]
+
+register_loader('equal', load_assign)
+```
+
+Extensions can also be associated to previously registered formats by adding them to the extension
+register as alias so it automatically detects with `.eq` extension:
+
+```python
+register_extension('eq', format='equal')
+```
+
 ## Parsing
 
 Given a path to an existing configuration file, it will be loaded in memory
@@ -151,7 +174,7 @@ confight.parse('/path/to/config', format='toml')
 When no format is given, it tries to guess by looking at file extensions:
 
 ```
-confight.parse('/path/to/config.json')  # will gess json format
+confight.parse('/path/to/config.json')  # will guess json format
 ```
 
 You can see the list of all available extensions at `confight.FORMAT_EXTENSIONS`.
@@ -371,7 +394,7 @@ Changelog
 * 1.4.0 (2023-12-12)
 
   [ Federico Fapitalle ]
-  * [3e618f3b] feat: adds support for HCL languaje
+  * [3e618f3b] feat: adds support for HCL language
 
   [ Frank Lenormand ]
   * [a9b3b9a2] fix(confight): Stick to older `ruamel.yaml` API
@@ -465,4 +488,3 @@ Changelog
 * 0.0.1 (2018-03-27)
 
   * Initial release.
-

confight.py

@@ -127,11 +127,10 @@ def parse(path, format=None):
     """
     format = format_from_path(path) if format is None else format
     logger.info("Parsing %r config file from %r", format, path)
-    if format not in FORMATS:
+    if format not in FORMAT_LOADERS:
         raise ValueError("Unknown format {} for file {}".format(format, path))
     loader = FORMAT_LOADERS[format]
-    with io.open(path, "r", encoding="utf8") as stream:
-        return loader(stream)
+    return loader(path)
 
 
 def merge(configs):
@@ -191,28 +190,62 @@ def check_access(path):
     return True
 
 
-def load_ini(stream):
+def open_text(path):
+    return io.open(path, mode="r", encoding="utf8")
+
+
+def open_bin(path):
+    return io.open(path, mode="rb", encoding="utf8")
+
+
+def load_json(path):
+    with open_text(path) as stream:
+        return json.load(stream, object_pairs_hook=OrderedDict)
+
+
+def load_toml(path):
+    with open_text(path) as stream:
+        return toml.load(stream, _dict=OrderedDict)
+
+
+def load_ini(path):
     if "ExtendedInterpolation" in globals():
         parser = ConfigParser(interpolation=ExtendedInterpolation())
     else:
         parser = ConfigParser()
-    parser.read_file(stream)
+    with open_text(path) as stream:
+        parser.read_file(stream)
     return {section: OrderedDict(parser.items(section)) for section in parser.sections()}
 
 
-FORMATS = ("toml", "ini", "json")
+FORMAT_LOADERS = {"json": load_json, "toml": load_toml, "ini": load_ini}
 FORMAT_EXTENSIONS = {
     "js": "json",
     "json": "json",
     "toml": "toml",
     "ini": "ini",
     "cfg": "ini",
 }
-FORMAT_LOADERS = {
-    "json": lambda *args: json.load(*args, object_pairs_hook=OrderedDict),
-    "toml": lambda *args: toml.load(*args, _dict=OrderedDict),
-    "ini": load_ini,
-}
+
+
+def register_format(name, function, override=False, _loaders=FORMAT_LOADERS):
+    if not override and name in _loaders:
+        raise ValueError("Format already registered: {}".format(name))
+    _loaders[name] = function
+
+
+def register_extension(
+    alias,
+    format,
+    override=False,
+    _loaders=FORMAT_LOADERS,
+    _extensions=FORMAT_EXTENSIONS,
+):
+    if format not in _loaders:
+        raise ValueError("Format does not exists: {}".format(format))
+    if not override and alias in _extensions:
+        raise ValueError("Format extension already registered: {}".format(alias))
+    _extensions[alias] = format
 
 
 # Optional dependency yaml
@@ -222,13 +255,14 @@ def load_ini(stream):
     pass
 else:
 
-    def load_yaml(stream):
-        yaml = YAML(typ="rt")
-        return yaml.load(stream)
+    def load_yaml(path):
+        with open_text(path) as stream:
+            yaml = YAML(typ="rt")
+            return yaml.load(stream)
 
-    FORMATS = FORMATS + ("yaml",)
-    FORMAT_EXTENSIONS.update({"yml": "yaml", "yaml": "yaml"})
-    FORMAT_LOADERS.update({"yaml": load_yaml})
+    register_format("yaml", load_yaml)
+    register_extension("yaml", format="yaml")
+    register_extension("yml", format="yaml")
 
 # Optional dependency HCL
 try:
@@ -237,12 +271,12 @@ def load_yaml(stream):
     pass
 else:
 
-    def load_hcl(stream):
-        return hcl.load(stream)
+    def load_hcl(path):
+        with open_text(path) as stream:
+            return hcl.load(stream)
 
-    FORMATS = FORMATS + ("hcl",)
-    FORMAT_EXTENSIONS.update({"hcl": "hcl"})
-    FORMAT_LOADERS.update({"hcl": load_hcl})
+    register_format("hcl", load_hcl)
+    register_extension("hcl", format="hcl")
 
 
 def format_from_path(path):
@@ -276,7 +310,11 @@ def cli():
     parser = argparse.ArgumentParser(description="One simple way of parsing configs")
     parser.add_argument("--version", action="version", version=get_version())
     parser.add_argument(
-        "-v", "--verbose", choices=LOG_LEVELS, default="ERROR", help="Logging level default: ERROR"
+        "-v",
+        "--verbose",
+        choices=LOG_LEVELS,
+        default="ERROR",
+        help="Logging level default: ERROR",
     )
     subparsers = parser.add_subparsers(title="subcommands", dest="command")
     show_parser = subparsers.add_parser("show")

dev-requirements.txt

@@ -1,3 +1,4 @@
 black
+black
 pyHamcrest
 pytest

test_confight.py

@@ -24,14 +24,16 @@
 )
 
 from confight import (
-    FORMATS,
+    FORMAT_LOADERS,
     find,
     load,
     load_app,
     load_paths,
     load_user_app,
     merge,
     parse,
+    register_extension,
+    register_format,
 )
 
 
@@ -40,18 +42,30 @@ def examples(tmpdir):
     return Repository(tmpdir)
 
 
-FILES = ["basic_file.toml", "basic_file.ini", "basic_file.json", "basic_file.cfg", "basic_file.js"]
-if "yaml" in FORMATS:
+FILES = [
+    "basic_file.toml",
+    "basic_file.ini",
+    "basic_file.json",
+    "basic_file.cfg",
+    "basic_file.js",
+]
+if "yaml" in FORMAT_LOADERS:
     FILES.extend(["basic_file.yaml", "basic_file.yml"])
-if "hcl" in FORMATS:
+if "hcl" in FORMAT_LOADERS:
     FILES.extend(["basic_file.hcl"])
 
-INVALID_FILES = ["invalid.toml", "invalid.ini", "invalid.json", "invalid.cfg", "invalid.js"]
+INVALID_FILES = [
+    "invalid.toml",
+    "invalid.ini",
+    "invalid.json",
+    "invalid.cfg",
+    "invalid.js",
+]
 INVALID_EXTENSIONS = ["bad_ext.ext", "bad_ext.j"]
 SORTED_FILES = ["00_base.toml", "01_first.json", "AA_second.ini"]
 
 
-class TestParse(object):
+class TestParse:
     @pytest.mark.parametrize("name", FILES)
     def test_it_should_detect_format_from_extension(self, name, examples):
         config = parse(examples.get(name))
@@ -92,7 +106,7 @@ def test_it_should_fail_with_invalid_extensions(self, name, examples):
             parse(examples.get(name))
 
 
-class TestMerge(object):
+class TestMerge:
     def test_it_should_give_priority_to_last_value(self):
         configs = [
             {"key": 1},
@@ -134,7 +148,8 @@ def test_it_should_merge_dicts_recursively(self):
         result = merge(configs)
 
         assert_that(
-            result, has_entry("section", has_entry("lv1", has_entry("lv2", has_entry("lv3", 3))))
+            result,
+            has_entry("section", has_entry("lv1", has_entry("lv2", has_entry("lv3", 3)))),
         )
 
     def test_it_should_ignore_scalar_values_given_as_configs(self):
@@ -151,7 +166,7 @@ def test_it_should_ignore_scalar_values_given_as_configs(self):
         assert_that(result, has_entry("section", has_entry("key", 3)))
 
 
-class TestFind(object):
+class TestFind:
     def test_it_should_load_files_in_order(self, examples):
         examples.clear()
         expected_files = sorted(examples.get_many(SORTED_FILES))
@@ -226,7 +241,7 @@ def test_it_should_warn_about_executable_config_files(self, logger, examples):
         logger.warning.assert_called()
 
 
-class TestLoad(object):
+class TestLoad:
     def test_it_should_load_and_merge_lists_of_paths(self, examples):
         paths = sorted(examples.get_many(SORTED_FILES))
 
@@ -262,7 +277,7 @@ def mymerge(configs):
         assert_that(config, only_contains(has_key("section")))
 
 
-class TestLoadPaths(object):
+class TestLoadPaths:
     def test_it_should_load_from_file_and_directory(self, examples):
         examples.clear()
         paths = sorted(examples.get_many(SORTED_FILES))
@@ -302,7 +317,7 @@ def test_merges_must_retain_order(self, examples):
         assert_that(config["section"].keys(), contains_exactly(*good_data))
 
 
-class LoadAppBehaviour(object):
+class LoadAppBehaviour:
     def loaded_paths(self, config):
         return sorted(config, key=lambda k: config[k])
 
@@ -506,7 +521,7 @@ def load_app(self, *args, **kwargs):
         return self.call_config_loader(load_user_app, *args, **kwargs)
 
 
-class TestCli(object):
+class TestCli:
     def test_it_should_print_help(self):
         out = subprocess.run([self.bin], stderr=subprocess.PIPE)
 
@@ -569,7 +584,7 @@ def maimed_run(*args, **kwargs):
     subprocess.run = maimed_run
 
 
-class Repository(object):
+class Repository:
     def __init__(self, tmpdir):
         self.tmpdir = tmpdir
 
@@ -696,3 +711,64 @@ def load(self, name):
     _contents["basic_file_yaml"] = _contents["basic_file.yaml"]
     _contents["basic_file.yml"] = _contents["basic_file.yaml"]
     _contents["basic_file_hcl"] = _contents["basic_file.hcl"]
+
+
+def function(path):
+    return {}
+
+
+def function2(path):
+    return {}
+
+
+class TestRegister:
+    def test_it_should_add_a_new_format(self):
+        register = {}
+
+        register_format("test", function, _loaders=register)
+
+        assert_that(register, has_entry("test", function))
+
+    def test_it_should_fail_with_existing_formats(self):
+        register = {"test": function}
+
+        with pytest.raises(ValueError):
+            register_format("test", function, _loaders=register)
+
+    def test_it_should_override_existing_formats(self):
+        register = {"test": function}
+
+        register_format("test", function2, override=True, _loaders=register)
+
+        assert_that(register, has_entry("test", function2))
+
+    def test_it_should_register_new_extensions(self):
+        loaders = {"test": function}
+        extensions = {}
+
+        register_extension("t", format="test", _loaders=loaders, _extensions=extensions)
+
+        assert_that(extensions, has_entry("t", "test"))
+
+    def test_it_should_fail_to_register_missing_formats(self):
+        loaders = {}
+
+        with pytest.raises(ValueError):
+            register_extension("t", format="test", _loaders=loaders, _extensions={})
+
+    def test_it_should_fail_with_existing_aliases(self):
+        loaders = {"test": lambda path: {}}
+        extensions = {"t": "test"}
+
+        with pytest.raises(ValueError):
+            register_extension("t", format="test", _loaders=loaders, _extensions=extensions)
+
+    def test_it_should_override_existing_aliases(self):
+        loaders = {"test1": lambda path: {}, "test2": lambda path: {}}
+        extensions = {"t": "test"}
+
+        register_extension(
+            "t", format="test2", override=True, _loaders=loaders, _extensions=extensions
+        )
+
+        assert_that(extensions, has_entry("t", "test2"))

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = py{38,39,310,311}-{basic,yaml,hcl}
+envlist = py{38,39,310,311,312}-{basic,yaml,hcl}
 skip_missing_interpreters=True
 
 [testenv]