Extendable ConfigLoader class

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "powerchord"
-version = "0.1.3"
+version = "0.1.4"
 description = "Concurrent CLI task runner"
 authors = ["Huib Piguillet <huib@proton.me>"]
 maintainers = ["Huib Piguillet"]

src/powerchord/__init__.py

@@ -0,0 +1,17 @@
+from .config import CLIConfig, Config, ConfigLoader, LoadConfigError, PyprojectConfig, load_config
+from .logging import LogLevel, LogLevels, logging_context
+from .runner import Task, TaskRunner
+
+__all__ = [
+    'Config',
+    'ConfigLoader',
+    'CLIConfig',
+    'LoadConfigError',
+    'LogLevel',
+    'LogLevels',
+    'PyprojectConfig',
+    'Task',
+    'TaskRunner',
+    'load_config',
+    'logging_context',
+]

src/powerchord/cli.py

@@ -1,7 +1,7 @@
 import asyncio
 import sys
 
-from .config import LoadConfigError, load_config
+from .config import CLIConfig, LoadConfigError, PyprojectConfig, load_config
 from .logging import logging_context
 from .runner import TaskRunner
 from .utils import catch_unknown_errors, killed_by
@@ -10,7 +10,7 @@
 @catch_unknown_errors()
 @killed_by(LoadConfigError)
 def main() -> None:
-    config = load_config()
+    config = load_config(CLIConfig, PyprojectConfig)
     with logging_context(config.log_levels):
         success = asyncio.run(TaskRunner(config.tasks).run_tasks())
     sys.exit(not success)

src/powerchord/config.py

@@ -1,8 +1,9 @@
 import argparse
 import tomllib
-from collections.abc import Callable
+from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from pathlib import Path
+from typing import ClassVar
 
 from chili import decode
 from gaffe import raises
@@ -15,6 +16,55 @@ class ParseConfigError(Exception):
     pass
 
 
+@dataclass
+class Config:
+    tasks: list[Task] = field(default_factory=list)
+    log_levels: LogLevels = field(default_factory=LogLevels)
+
+
+class DecodeConfigError(Exception):
+    pass
+
+
+@dataclass
+class ConfigLoader(ABC):
+    name: ClassVar[str]
+
+    tasks: list[Task] = field(default_factory=list)
+    log_levels: LogLevels = field(default_factory=LogLevels)
+
+    @classmethod
+    @raises(ParseConfigError)
+    @abstractmethod
+    def _parse(cls) -> dict:
+        pass
+
+    @classmethod
+    @raises(ParseConfigError, DecodeConfigError)
+    def load(cls) -> Config | None:
+        config_dict = cls._parse()
+        if not any(config_dict.values()):
+            return None
+
+        tasks = config_dict.get('tasks', {})
+        if isinstance(tasks, list):
+            task_items = [('', t) if isinstance(t, str) else t for t in tasks]
+        elif isinstance(tasks, dict):
+            task_items = list(tasks.items())
+        else:
+            raise DecodeConfigError(f'Wrong value for tasks: {tasks}')
+        config_dict['tasks'] = [{'command': t, 'name': n} for n, t in task_items]
+
+        log_levels = config_dict.get('log_levels', {})
+        if isinstance(log_levels, list):
+            config_dict['log_levels'] = dict(log_levels)
+
+        try:
+            return decode(config_dict, Config, decoders={LogLevel: LogLevel})
+        except ValueError as exc:
+            raise DecodeConfigError(*exc.args) from exc
+
+
 def parse_key_value_pair(value: str) -> tuple[str, str]:
     key, value = value.split('=', 1)
     return key, value
@@ -27,75 +77,50 @@ def try_parse_key_value_pair(value: str) -> str | tuple[str, str]:
         return value
 
 
-@raises(ParseConfigError)
-def config_from_args(_config_source: str) -> dict:
-    arg_parser = argparse.ArgumentParser()
-    arg_parser.add_argument(
-        '-t',
-        '--tasks',
-        dest='tasks',
-        nargs='+',
-        metavar='COMMAND | NAME=COMMAND',
-        type=try_parse_key_value_pair,
-        default={},
-    )
-    arg_parser.add_argument(
-        '-l',
-        '--log-levels',
-        dest='log_levels',
-        nargs='+',
-        metavar='OUTPUT=LOGLEVEL (debug | info | warning | error | critical | "")',
-        type=parse_key_value_pair,
-        default={},
-    )
-    try:
-        return arg_parser.parse_args().__dict__
-    except (SystemExit, TypeError) as exc:
-        raise ParseConfigError from exc
-
-
-@raises(ParseConfigError)
-def config_from_pyproject(config_source: str) -> dict:
-    try:
-        with Path(config_source).open('rb') as f:
-            return tomllib.load(f).get('tool', {}).get('powerchord', {})
-    except OSError:
-        return {}
-    except ValueError as exc:
-        raise ParseConfigError from exc
-
-
-@dataclass
-class Config:
-    tasks: list[Task] = field(default_factory=list)
-    log_levels: LogLevels = field(default_factory=LogLevels)
-
-
-class DecodeConfigError(Exception):
-    pass
-
+class CLIConfig(ConfigLoader):
+    name = 'command line'
+
+    @classmethod
+    @raises(ParseConfigError)
+    def _parse(cls) -> dict:
+        arg_parser = argparse.ArgumentParser()
+        arg_parser.add_argument(
+            '-t',
+            '--tasks',
+            dest='tasks',
+            nargs='+',
+            metavar='COMMAND | NAME=COMMAND',
+            type=try_parse_key_value_pair,
+            default={},
+        )
+        arg_parser.add_argument(
+            '-l',
+            '--log-levels',
+            dest='log_levels',
+            nargs='+',
+            metavar='OUTPUT=LOGLEVEL (debug | info | warning | error | critical | "")',
+            type=parse_key_value_pair,
+            default={},
+        )
+        try:
+            return arg_parser.parse_args().__dict__
+        except (argparse.ArgumentError, SystemExit, TypeError) as exc:
+            raise ParseConfigError from exc
 
-@raises(DecodeConfigError)
-def decode_config(value: dict) -> Config | None:
-    if not any(value.values()):
-        return None
-    tasks = value.get('tasks', {})
-    if isinstance(tasks, list):
-        task_items = [('', t) if isinstance(t, str) else t for t in tasks]
-    elif isinstance(tasks, dict):
-        task_items = list(tasks.items())
-    else:
-        raise DecodeConfigError(f'Wrong value for tasks: {tasks}')
-    value['tasks'] = [{'command': t, 'name': n} for n, t in task_items]
 
-    log_levels = value.get('log_levels', {})
-    if isinstance(log_levels, list):
-        value['log_levels'] = dict(log_levels)
+class PyprojectConfig(ConfigLoader):
+    name = 'pyproject.toml'
 
-    try:
-        return decode(value, Config, decoders={LogLevel: LogLevel})
-    except ValueError as exc:
-        raise DecodeConfigError(*exc.args) from exc
+    @classmethod
+    @raises(ParseConfigError)
+    def _parse(cls) -> dict:
+        try:
+            with Path('pyproject.toml').open('rb') as f:
+                return tomllib.load(f).get('tool', {}).get('powerchord', {})
+        except OSError:
+            return {}
+        except ValueError as exc:
+            raise ParseConfigError from exc
 
 
 class LoadConfigError(Exception):
@@ -108,19 +133,13 @@ def __init__(self, name: str, *args):
         super().__init__(f' from {name}', *args)
 
 
-CONFIG_LOADERS: dict[str, Callable[[str], dict]] = {
-    'command line': config_from_args,
-    'pyproject.toml': config_from_pyproject,
-}
-
-
 @raises(LoadConfigError)
-def load_config() -> Config:
-    for name, loader in CONFIG_LOADERS.items():
+def load_config(*loaders: type[ConfigLoader]) -> Config:
+    for loader_cls in loaders:
         try:
-            config = decode_config(loader(name))
-        except ParseConfigError as exc:
-            raise LoadSpecificConfigError(name, *exc.args) from exc
+            config = loader_cls.load()
+        except (ParseConfigError, DecodeConfigError) as exc:
+            raise LoadSpecificConfigError(loader_cls.name, *exc.args) from exc
         if config:
             return config
     raise LoadConfigError