Merge branch 'patch-config' into 'v3'

Specify config patches rather than full config files in settings

See merge request openflexure/openflexure-microscope-server!420

src/openflexure_microscope_server/server/__init__.py

@@ -6,10 +6,13 @@ from typing import Optional, Callable, Any
 from functools import wraps
 from copy import copy
 import logging
+from argparse import Namespace
 
 import labthings_fastapi as lt
 import uvicorn
 from uvicorn.main import Server
+
+from openflexure_microscope_server.utilities import load_patched_config
 from .serve_static_files import add_static_files
 from .legacy_api import add_v2_endpoints
 from ..logging import configure_logging, retrieve_log, retrieve_log_from_file
@@ -81,7 +84,7 @@ def serve_from_cli(argv: Optional[list[str]] = None) -> None:
     config = None
     server = None
     try:
-        config = lt.cli.config_from_args(args)
+        config = _full_config_from_args(args)
         log_folder = config.get("log_folder", "./openflexure/logs")
         scans_folder = _get_scans_dir(config)
         server = lt.cli.server_from_config(config)
@@ -128,3 +131,16 @@ def serve_from_cli(argv: Optional[list[str]] = None) -> None:
             )
         else:
             raise e
+
+
+def _full_config_from_args(args: Namespace) -> dict:
+    """Load configuration from LabThings args allowing patching.
+
+    This provides similar functionarlity to lt.cli.config_from_args except allows the
+    configuration file to specify a base config, and optionally patches.
+    """
+    # If no config file specified let LabThings handle it.
+    if not args.config:
+        return lt.cli.config_from_args(args)
+
+    return load_patched_config(args.config)

src/openflexure_microscope_server/utilities.py

@@ -9,6 +9,8 @@ from typing import (
     Concatenate,
     Self,
     overload,
+    TypeAlias,
+    Literal,
 )
 import os
 import re
@@ -18,6 +20,7 @@ import logging
 from importlib.metadata import version
 import tomllib
 from functools import wraps
+import json
 
 from pydantic import BaseModel
 import numpy as np
@@ -25,6 +28,9 @@ import numpy as np
 T = TypeVar("T")
 P = ParamSpec("P")
 
+JSONScalar: TypeAlias = str | int | float | bool | None
+JSONType: TypeAlias = dict[str, "JSONType"] | list["JSONType"] | JSONScalar
+
 LOGGER = logging.getLogger(__name__)
 
 REPO_DIR = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
@@ -359,3 +365,136 @@ def quadratic(
     :return: The quadratic, evaluated at each point in ``x``
     """
     return a * x**2 + b * x + c
+
+
+# Use overload to clarify to MyPy that if enforce_dict is true, inputs and outputs are
+# dictionaries
+@overload
+def merge_patch(
+    target: dict[str, JSONType], patch: dict[str, JSONType], enforce_dict: Literal[True]
+) -> dict[str, JSONType]: ...
+
+
+# If not they are any JSONType
+@overload
+def merge_patch(
+    target: JSONType, patch: JSONType, enforce_dict: Literal[False]
+) -> JSONType: ...
+@overload
+def merge_patch(target: JSONType, patch: JSONType) -> JSONType: ...
+
+
+def merge_patch(
+    target: JSONType, patch: JSONType, enforce_dict: bool = False
+) -> JSONType:
+    """Merge json data using the methods from IETF RFC 7396.
+
+    Primarily this is designed to merge dictionaries, but IETF RFC 7396 provides
+    defined methods for handling non-dictionary data loaded from JSON, so this
+    has been provided in full.
+
+    :param target: The target object
+    :param patch: The patch to be applied
+    :param enforce_dict: Boolean, set True enfoces that the target and patch are both
+        dictionaries.
+    """
+    if enforce_dict and not (isinstance(target, dict) and isinstance(patch, dict)):
+        raise ValueError("Both target and patch should be dictionaries.")
+
+    # If patch is not a dict (object), it replaces target entirely:
+    if not isinstance(patch, dict):
+        return patch
+
+    # If target is not an object, replace it with an empty object
+    if not isinstance(target, dict):
+        target = {}
+
+    result = {}
+    # First, keep all keys from target that are not in patch
+    for key, target_val in target.items():
+        if key not in patch:
+            result[key] = target_val
+
+    # Then apply patch keys
+    for key, patch_val in patch.items():
+        # If patch_val is None, do not add the key (deleting from target).
+        if patch_val is None:
+            continue
+
+        # Check if values are dictionaries.
+        if isinstance(patch_val, dict):
+            # If patch value is a dictionary then recurse.
+            result[key] = merge_patch(target.get(key), patch_val)
+        else:
+            # Else, use the patch value
+            result[key] = patch_val
+    return result
+
+
+def load_patched_config(config_path: str) -> dict:
+    """Load a json configuration file, if it a patch, return the full config.
+
+    Load a json file. If it does not contains the "base_config_file" key then return
+    the data as read.
+
+    If the configuration specifies a "base_config_file" but no "patch" then return the
+    data from reading base_config_file as json.
+
+    If the configuration specifies a "base_config_file" and "patch" then apply the
+    patch to the data in base_config_file and return the result. The patching method
+    is merge_patch()
+
+    :param config_path: The path of the configuration file.
+    :return: The contents of the configuration file after loading and patching the
+        specified base configuration file if applicable.
+
+    """
+    try:
+        with open(config_path, "r", encoding="utf-8") as f:
+            config = json.load(f)
+    except IOError as e:
+        raise type(e)(f"Couldn't load configuration file {config_path}") from e
+    except json.JSONDecodeError as e:
+        raise json.JSONDecodeError(
+            f"Invalid JSON in configuration file {config_path}: {e}", e.doc, e.pos
+        ) from e
+
+    # If base_config_file not specified then this is a normal LabThings config file
+    # Just return it.
+    if "base_config_file" not in config:
+        return config
+
+    config_dir = os.path.dirname(os.path.abspath(config_path))
+    base_conf_path = config["base_config_file"]
+
+    base_conf_path = resolve_path_from_dir(base_conf_path, config_dir)
+
+    try:
+        with open(base_conf_path, "r", encoding="utf-8") as f:
+            base_config = json.load(f)
+    except IOError as e:
+        raise type(e)(f"Couldn't load base configuration file {base_conf_path}") from e
+    except json.JSONDecodeError as e:
+        raise json.JSONDecodeError(
+            f"Invalid JSON in base configuration file {config_path}: {e}", e.doc, e.pos
+        ) from e
+
+    if "patch" in config:
+        return merge_patch(base_config, config["patch"], enforce_dict=True)
+
+    return base_config
+
+
+def resolve_path_from_dir(path: str, directory: str) -> str:
+    """Convert a relative or abs path specified in one dir to the working dir.
+
+    This also expands any environment variables.
+
+    :param path: The specified path (could be absolute or relative)
+    :param directory: The directory from which the path was specified
+    :return: The normalised path.
+    """
+    path = os.path.expandvars(os.path.expanduser(path))
+    if not os.path.isabs(path):
+        path = os.path.join(directory, path)
+    return os.path.normpath(path)

tests/test_config_utilities.py

@@ -0,0 +1,124 @@
+"""Test the configuration handling functions in utilities."""
+
+import pytest
+import os
+import json
+from json import JSONDecodeError
+
+from openflexure_microscope_server import utilities
+
+
+@pytest.mark.parametrize(
+    ("target", "patch", "outcome", "err_if_enforce"),
+    [
+        ({"a": "b"}, {"a": "c"}, {"a": "c"}, False),
+        ({"a": "b"}, {"b": "c"}, {"a": "b", "b": "c"}, False),
+        ({"a": "b"}, {"a": None}, {}, False),
+        ({"a": "b", "b": "c"}, {"a": None}, {"b": "c"}, False),
+        ({"a": ["b"]}, {"a": "c"}, {"a": "c"}, False),
+        ({"a": "c"}, {"a": ["b"]}, {"a": ["b"]}, False),
+        (
+            {"a": {"b": "c"}},
+            {"a": {"b": "d", "c": None}},
+            {"a": {"b": "d"}},
+            False,
+        ),
+        ({"a": [{"b": "c"}]}, {"a": [1]}, {"a": [1]}, False),
+        (["a", "b"], ["c", "d"], ["c", "d"], True),
+        ({"a": "b"}, ["c"], ["c"], True),
+        ({"a": "foo"}, None, None, True),
+        ({"a": "foo"}, "bar", "bar", True),
+        ({"e": None}, {"a": 1}, {"e": None, "a": 1}, False),
+        ([1, 2], {"a": "b", "c": None}, {"a": "b"}, True),
+        ({}, {"a": {"bb": {"ccc": None}}}, {"a": {"bb": {}}}, False),
+    ],
+)
+def test_merge_patch(target, patch, outcome, err_if_enforce):
+    """Check that merge_patch returns the expected outcome for each target and patch.
+
+    These test cases are specified in Appedix A of IETF RFC 7396.
+    """
+    assert utilities.merge_patch(target, patch) == outcome
+
+    # If expected to error when enforce_dict is True then check:
+    if err_if_enforce:
+        with pytest.raises(ValueError, match="Both target and patch"):
+            utilities.merge_patch(target, patch, enforce_dict=True)
+    else:
+        # Else should run without error with same expected value
+        assert utilities.merge_patch(target, patch, enforce_dict=True) == outcome
+
+
+@pytest.mark.parametrize(
+    ("base_conf", "resolves_to"),
+    [
+        ("/var/base.json", "/var/base.json"),
+        ("base.json", "/var/openflexure/settings/base.json"),
+        ("./base.json", "/var/openflexure/settings/base.json"),
+        ("../base.json", "/var/openflexure/base.json"),
+        (
+            "$OFM_LIB/base.json",
+            "/var/openflexure/application/openflexure-microscope-server/base.json",
+        ),
+    ],
+)
+def test_resolve_path_from_dir(base_conf, resolves_to, mocker):
+    """Test that resolve path handles, absolute and relative paths and expands env vars."""
+    mocker.patch.dict(
+        os.environ,
+        {"OFM_LIB": "/var/openflexure/application/openflexure-microscope-server"},
+    )
+    conf_dir = "/var/openflexure/settings"
+    resolved_path = utilities.resolve_path_from_dir(base_conf, directory=conf_dir)
+    assert resolved_path == resolves_to
+
+
+def test_patching_config(tmpdir):
+    """Test each situation in load_patched_config."""
+    conf_file = os.path.join(tmpdir, "ofm_config.json")
+    base_conf_file = os.path.join(tmpdir, "ofm_base_config.json")
+
+    simple_config = {"things": {"thing1": "python.class"}}
+    config_patch = {"things": {"thing2": "python.class2"}}
+    patched_config = {"things": {"thing1": "python.class", "thing2": "python.class2"}}
+    broken_json = "{{invalid{Json"
+
+    # Error if config file doesn't exist
+    with pytest.raises(IOError, match="Couldn't load configuration"):
+        utilities.load_patched_config(conf_file)
+
+    # Error if config file contains bad json
+    with open(conf_file, "w", encoding="utf-8") as file_obj:
+        file_obj.write(broken_json)
+    with pytest.raises(JSONDecodeError, match="Invalid JSON in configuration"):
+        utilities.load_patched_config(conf_file)
+
+    # If base_config_file is not set, just return the configuration
+    with open(conf_file, "w", encoding="utf-8") as file_obj:
+        json.dump(simple_config, file_obj)
+    assert utilities.load_patched_config(conf_file) == simple_config
+
+    # Set the "base_config_file" instead, error as there is no base config file
+    config = {"base_config_file": "./ofm_base_config.json"}
+    with open(conf_file, "w", encoding="utf-8") as file_obj:
+        json.dump(config, file_obj)
+    with pytest.raises(IOError, match="Couldn't load base configuration"):
+        utilities.load_patched_config(conf_file)
+
+    # Error if base config file contains bad json
+    with open(base_conf_file, "w", encoding="utf-8") as file_obj:
+        file_obj.write(broken_json)
+    with pytest.raises(JSONDecodeError, match="Invalid JSON in base configuration"):
+        utilities.load_patched_config(conf_file)
+
+    # Loads the simple configuration from the base config file.
+    with open(base_conf_file, "w", encoding="utf-8") as file_obj:
+        json.dump(simple_config, file_obj)
+    assert utilities.load_patched_config(conf_file) == simple_config
+
+    # Set the outer config to have a patch.
+    # It now reads the simple_config from the base condif file and patches it
+    config["patch"] = config_patch
+    with open(conf_file, "w", encoding="utf-8") as file_obj:
+        json.dump(config, file_obj)
+    assert utilities.load_patched_config(conf_file) == patched_config