Merge pull request #54 from tangkong/enh_config_client

ENH: Add ability to instantiate Client from configuration file

docs/source/upcoming_release_notes/54-enh_config_client.rst

@@ -0,0 +1,22 @@
+54 enh_config_client
+####################
+
+API Breaks
+----------
+- N/A
+
+Features
+--------
+- Adds ability for Client to search for configuration files, and load settings from them.
+
+Bugfixes
+--------
+- N/A
+
+Maintenance
+-----------
+- N/A
+
+Contributors
+------------
+- tangkong

superscore/backends/__init__.py

@@ -0,0 +1,58 @@
+__all__ = ['BACKENDS']
+
+import logging
+from typing import Dict
+
+from .core import _Backend
+
+logger = logging.getLogger(__name__)
+
+
+BACKENDS: Dict[str, _Backend] = {}
+
+
+def _get_backend(backend: str) -> _Backend:
+    if backend == 'filestore':
+        from .filestore import FilestoreBackend
+        return FilestoreBackend
+    if backend == 'test':
+        from .test import TestBackend
+        return TestBackend
+
+    raise ValueError(f"Unknown backend {backend}")
+
+
+def _init_backends() -> Dict[str, _Backend]:
+    backends = {}
+
+    try:
+        backends['filestore'] = _get_backend('filestore')
+    except ImportError as ex:
+        logger.debug(f"Filestore Backend unavailable: {ex}")
+
+    try:
+        backends['test'] = _get_backend('test')
+    except ImportError as ex:
+        logger.debug(f"Test Backend unavailable: {ex}")
+
+    return backends
+
+
+def get_backend(backend_name: str) -> _Backend:
+    try:
+        backend = BACKENDS[backend_name]
+    except KeyError:
+        # try to load it
+        try:
+            BACKENDS[backend_name] = _get_backend(backend_name)
+            backend = BACKENDS[backend_name]
+        except ValueError:
+            raise ValueError(f'Backend {backend_name} not supported. Available '
+                             f'backends include: ({list(BACKENDS.keys())})')
+        except ImportError as ex:
+            raise ValueError(f'Backend {(backend_name)} failed to load: {ex}')
+
+    return backend
+
+
+BACKENDS = _init_backends()

superscore/backends/test.py

@@ -1,7 +1,7 @@
 """
 Backend that manipulates Entries in-memory for testing purposes.
 """
-from typing import List
+from typing import List, Optional
 from uuid import UUID
 
 from superscore.backends.core import _Backend
@@ -12,8 +12,11 @@
 
 class TestBackend(_Backend):
     """Backend that manipulates Entries in-memory, for testing purposes."""
-    def __init__(self, data: List[Entry]):
-        self.data = data
+    def __init__(self, data: Optional[List[Entry]] = None):
+        if data is None:
+            self.data = []
+        else:
+            self.data = data
 
     def save_entry(self, entry: Entry) -> None:
         try:

superscore/client.py

@@ -1,8 +1,12 @@
 """Client for superscore.  Used for programmatic interactions with superscore"""
+import configparser
 import logging
+import os
+from pathlib import Path
 from typing import Any, Generator, List, Optional, Union
 from uuid import UUID
 
+from superscore.backends import get_backend
 from superscore.backends.core import _Backend
 from superscore.control_layers import ControlLayer
 from superscore.control_layers.status import TaskStatus
@@ -15,13 +19,128 @@ class Client:
     backend: _Backend
     cl: ControlLayer
 
-    def __init__(self, backend: _Backend, **kwargs) -> None:
+    def __init__(
+        self,
+        backend: Optional[_Backend] = None,
+        control_layer: Optional[ControlLayer] = None,
+    ) -> None:
+        if backend is None:
+            # set up a temp backend with temp file
+            logger.warning('No backend specified, loading an empty test backend')
+            backend = get_backend('test')()
+        if control_layer is None:
+            control_layer = ControlLayer()
+
         self.backend = backend
-        self.cl = ControlLayer()
+        self.cl = control_layer
 
     @classmethod
-    def from_config(cls, cfg=None):
-        raise NotImplementedError
+    def from_config(cls, cfg: Path = None):
+        """
+        Create a client from the configuration file specification.
+
+        Configuration file should be of an "ini" format, along the lines of:
+
+        .. code::
+
+            [backend]
+            type = filestore
+            path = ./db/filestore.json
+
+            [control_layer]
+            ca = true
+            pva = true
+
+        The ``backend`` section has one special key ("type"), and the rest of the
+        settings are passed to the appropriate ``_Backend`` as keyword arguments.
+
+        The ``control_layer`` section has a key-value pair for each available shim.
+        The ``ControlLayer`` object will be created with all the valid shims with
+        True values.
+
+        Parameters
+        ----------
+        cfg : Path, optional
+            Path to a configuration file, by default None.  If omitted,
+            :meth:`.find_config` will be used to find one
+
+        Raises
+        ------
+        RuntimeError
+            If a configuration file cannot be found
+        """
+        if not cfg:
+            cfg = cls.find_config()
+        if not os.path.exists(cfg):
+            raise RuntimeError(f"Superscore configuration file not found: {cfg}")
+
+        cfg_parser = configparser.ConfigParser()
+        cfg_file = cfg_parser.read(cfg)
+        logger.debug(f"Loading configuration file at ({cfg_file})")
+
+        # Gather Backend
+        if 'backend' in cfg_parser.sections():
+            backend_type = cfg_parser.get("backend", "type")
+            kwargs = {key: value for key, value
+                      in cfg_parser["backend"].items()
+                      if key != "type"}
+            backend_class = get_backend(backend_type)
+            backend = backend_class(**kwargs)
+        else:
+            logger.warning('No backend specified, loading an empty test backend')
+            backend = get_backend('test')()
+
+        # configure control layer and shims
+        if 'control_layer' in cfg_parser.sections():
+            shim_choices = [val for val, enabled
+                            in cfg_parser["control_layer"].items()
+                            if enabled]
+            control_layer = ControlLayer(shims=shim_choices)
+        else:
+            logger.debug('No control layer shims specified, loading all available')
+            control_layer = ControlLayer()
+
+        return cls(backend=backend, control_layer=control_layer)
+
+    @staticmethod
+    def find_config() -> Path:
+        """
+        Search for a ``superscore`` configuation file.  Searches in the following
+        locations in order
+        - ``$SUPERSCORE_CFG`` (a full path to a config file)
+        - ``$XDG_CONFIG_HOME/{superscore.cfg, .superscore.cfg}`` (either filename)
+        - ``~/.config/{superscore.cfg, .superscore.cfg}``
+
+        Returns
+        -------
+        path : str
+            Absolute path to the configuration file
+
+        Raises
+        ------
+        OSError
+            If no configuration file can be found by the described methodology
+        """
+        # Point to with an environment variable
+        if os.environ.get('SUPERSCORE_CFG', False):
+            superscore_cfg = os.environ.get('SUPERSCORE_CFG')
+            logger.debug("Found $SUPERSCORE_CFG specification for Client "
+                         "configuration at %s", superscore_cfg)
+            return superscore_cfg
+        # Search in the current directory and home directory
+        else:
+            config_dirs = [os.environ.get('XDG_CONFIG_HOME', "."),
+                           os.path.expanduser('~/.config'),]
+            for directory in config_dirs:
+                logger.debug('Searching for superscore config in %s', directory)
+                for path in ('.superscore.cfg', 'superscore.cfg'):
+                    full_path = os.path.join(directory, path)
+
+                    if os.path.exists(full_path):
+                        logger.debug("Found configuration file at %r", full_path)
+                        return full_path
+        # If found nothing
+        raise OSError("No superscore configuration file found. Check SUPERSCORE_CFG.")
 
     def search(self, **post) -> Generator[Entry, None, None]:
         """Search by key-value pair.  Can search by any field, including id"""

superscore/control_layers/core.py

@@ -3,14 +3,22 @@
 and dispatches to various shims depending on the context.
 """
 import asyncio
+import logging
 from functools import singledispatchmethod
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Dict, List, Optional, Union
 
 from superscore.control_layers.status import TaskStatus
 
 from ._aioca import AiocaShim
 from ._base_shim import _BaseShim
 
+logger = logging.getLogger(__name__)
+
+# available communication shim layers
+SHIMS = {
+    'ca': AiocaShim()
+}
+
 
 class ControlLayer:
     """
@@ -19,10 +27,16 @@ class ControlLayer:
     """
     shims: Dict[str, _BaseShim]
 
-    def __init__(self, *args, **kwargs):
-        self.shims = {
-            'ca': AiocaShim(),
-        }
+    def __init__(self, *args, shims: Optional[List[str]] = None, **kwargs):
+        if shims is None:
+            # load all available shims
+            self.shims = SHIMS
+            logger.debug('No shims specified, loading all available communication '
+                         f'shims: {list(self.shims.keys())}')
+        else:
+            self.shims = {key: shim for key, shim in SHIMS.items() if key in shims}
+            logger.debug('Loaded valid shims from the requested list: '
+                         f'{list(self.shims.keys())}')
 
     def shim_from_pv(self, address: str) -> _BaseShim:
         """

superscore/tests/config.cfg

@@ -0,0 +1,7 @@
+[backend]
+type = filestore
+path = ./db/filestore.json
+
+[control_layer]
+ca = true
+pva = true

superscore/tests/test_client.py

@@ -1,10 +1,43 @@
+import os
+from pathlib import Path
 from unittest.mock import patch
 
+import pytest
+
+from superscore.backends.filestore import FilestoreBackend
 from superscore.client import Client
 from superscore.model import Root
 
 from .conftest import MockTaskStatus
 
+SAMPLE_CFG = Path(__file__).parent / 'config.cfg'
+
+
+@pytest.fixture(scope='function')
+def xdg_config_patch(tmp_path):
+    config_home = tmp_path / 'xdg_config_home'
+    config_home.mkdir()
+    return config_home
+
+
+@pytest.fixture(scope='function')
+def sscore_cfg(xdg_config_patch: Path):
+    # patch config discovery paths
+    xdg_cfg = os.environ.get("XDG_CONFIG_HOME", '')
+    sscore_cfg = os.environ.get("SUPERSCORE_CFG", '')
+
+    os.environ['XDG_CONFIG_HOME'] = str(xdg_config_patch)
+    os.environ['SUPERSCORE_CFG'] = ''
+
+    sscore_cfg_path = xdg_config_patch / "superscore.cfg"
+    sscore_cfg_path.symlink_to(SAMPLE_CFG)
+
+    yield str(sscore_cfg_path)
+
+    # reset env vars
+    os.environ["SUPERSCORE_CFG"] = str(sscore_cfg)
+    os.environ["XDG_CONFIG_HOME"] = xdg_cfg
+
 
 @patch('superscore.control_layers.core.ControlLayer.put')
 def test_apply(put_mock, mock_client: Client, sample_database: Root):
@@ -19,3 +52,17 @@ def test_apply(put_mock, mock_client: Client, sample_database: Root):
 
     mock_client.apply(snap, sequential=True)
     assert put_mock.call_count == 3
+
+
+def test_from_cfg(sscore_cfg: str):
+    client = Client.from_config()
+    assert isinstance(client.backend, FilestoreBackend)
+    assert 'ca' in client.cl.shims
+
+
+def test_find_config(sscore_cfg: str):
+    assert sscore_cfg == Client.find_config()
+
+    # explicit SUPERSCORE_CFG env var supercedes XDG_CONFIG_HOME
+    os.environ['SUPERSCORE_CFG'] = 'other/cfg'
+    assert 'other/cfg' == Client.find_config()