create main get_configuration endpoint and add tests

Author: olliesilvester

Dockerfile

@@ -27,8 +27,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
 COPY --from=build /venv/ /venv/
 COPY tests/test_data/beamline_parameters.txt tests/test_data/beamline_parameters.txt
 ENV PATH=/venv/bin:$PATH
-ARG RUN_APP_IN_DEV_MODE=0
-ENV DEV_MODE=${RUN_APP_IN_DEV_MODE}
+ARG BEAMLINE="dev"
+ENV BEAMLINE=${RUN_APP_IN_DEV_MODE}
 
 # change this entrypoint if it is not the same as the repo
 CMD daq-config-server

src/daq_config_server/__main__.py

@@ -3,7 +3,6 @@
 from . import __version__
 
 try:
-    import redis  # noqa
     import uvicorn  # noqa
     from fastapi import FastAPI  # noqa
 
@@ -18,8 +17,7 @@
 def main():
     parser = ArgumentParser()
     parser.add_argument("-v", "--version", action="version", version=__version__)
-    parser.add_argument("-d", "--dev", action="store_true")
-    args = parser.parse_args()
+
     if not server_dependencies_exist:
         print(
             "To do anything other than print the version and be available for "
@@ -29,7 +27,7 @@ def main():
     else:
         from .app import main
 
-        main(args)
+        main()
 
 
 # test with: python -m daq_config_server

src/daq_config_server/app.py

@@ -1,31 +1,14 @@
-from os import environ
+from pathlib import Path
 
 import uvicorn
-from fastapi import FastAPI, Request, Response, status
+from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from pydantic import BaseModel
-from redis import Redis
 
-from .beamline_parameters import (
-    BEAMLINE_PARAMETER_PATHS,
-    GDABeamlineParameters,
-)
-from .constants import DATABASE_KEYS, ENDPOINTS
-
-DEV_MODE = bool(int(environ.get("DEV_MODE") or 0))
-
-ROOT_PATH = "/api"
-print(f"{DEV_MODE=}")
-print(f"{ROOT_PATH=}")
-if DEV_MODE:
-    print("Running in dev mode! not setting root path!")
-    ROOT_PATH = ""
+from daq_config_server.constants import ENDPOINTS
 
 app = FastAPI(
     title="DAQ config server",
-    description="""For storing and fetching beamline parameters, etc. which are needed
-    by more than one applicatioon or service""",
-    root_path=ROOT_PATH,
+    description="""For reading files stored on /dls_sw from another container""",
 )
 origins = ["*"]
 app.add_middleware(
@@ -36,124 +19,21 @@
     allow_headers=["*"],
 )
 
-valkey = Redis(host="localhost", port=6379, decode_responses=True)
-
 __all__ = ["main"]
 
-BEAMLINE_PARAM_PATH = ""
-BEAMLINE_PARAMS: GDABeamlineParameters | None = None
-
-
-@app.get(ENDPOINTS.BL_PARAM + "/{param}")
-def get_beamline_parameter(param: str):
-    """Get a single beamline parameter"""
-    assert BEAMLINE_PARAMS is not None
-    return {param: BEAMLINE_PARAMS.params.get(param)}
-
-
-class ParamList(BaseModel):
-    param_list: list[str]
-
-
-@app.get(ENDPOINTS.BL_PARAM)
-def get_all_beamline_parameters(param_list_data: ParamList | None = None):
-    """Get a dict of all the current beamline parameters."""
-    assert BEAMLINE_PARAMS is not None
-    if param_list_data is None:
-        return BEAMLINE_PARAMS.params
-    return {k: BEAMLINE_PARAMS.params.get(k) for k in param_list_data.param_list}
-
-
-@app.get(ENDPOINTS.FEATURE)
-def get_feature_flag_list(get_values: bool = False):
-    """Get a list of all the current feature flags, or a dict of all the current values
-    if get_values=true is passed"""
-    flags = valkey.smembers(DATABASE_KEYS.FEATURE_SET)
-    if not get_values:
-        return flags
-    else:
-        return {flag: bool(int(valkey.get(flag))) for flag in flags}  # type: ignore
-
-
-@app.get(ENDPOINTS.FEATURE + "/{flag_name}")
-def get_feature_flag(flag_name: str, response: Response):
-    """Get the value of a feature flag"""
-    if not valkey.sismember(DATABASE_KEYS.FEATURE_SET, flag_name):
-        response.status_code = status.HTTP_404_NOT_FOUND
-        return {"message": f"Feature flag {flag_name} does not exist!"}
-    else:
-        ret = int(valkey.get(flag_name))  # type: ignore # We checked if it exists above
-        return {flag_name: bool(ret)}
-
-
-@app.post(ENDPOINTS.FEATURE + "/{flag_name}", status_code=status.HTTP_201_CREATED)
-def create_feature_flag(flag_name: str, response: Response, value: bool = False):
-    """Sets a feature flag, creating it if it doesn't exist. Default to False."""
-    if valkey.sismember(DATABASE_KEYS.FEATURE_SET, flag_name):
-        response.status_code = status.HTTP_409_CONFLICT
-        return {"message": f"Feature flag {flag_name} already exists!"}
-    else:
-        valkey.sadd(DATABASE_KEYS.FEATURE_SET, flag_name)
-        return {"success": valkey.set(flag_name, int(value))}
-
-
-@app.put(ENDPOINTS.FEATURE + "/{flag_name}")
-def set_feature_flag(flag_name: str, value: bool, response: Response):
-    """Sets a feature flag, return an error if it doesn't exist."""
-    if not valkey.sismember(DATABASE_KEYS.FEATURE_SET, flag_name):
-        response.status_code = status.HTTP_404_NOT_FOUND
-        return {"message": f"Feature flag {flag_name} does not exist!"}
-    else:
-        return {"success": valkey.set(flag_name, int(value))}
-
-
-@app.delete(ENDPOINTS.FEATURE + "/{flag_name}")
-def delete_feature_flag(flag_name: str, response: Response):
-    """Delete a feature flag."""
-    if not valkey.sismember(DATABASE_KEYS.FEATURE_SET, flag_name):
-        response.status_code = status.HTTP_404_NOT_FOUND
-        return {"message": f"Feature flag {flag_name} does not exist!"}
-    else:
-        valkey.srem(DATABASE_KEYS.FEATURE_SET, flag_name)
-        return {"success": not valkey.sismember(DATABASE_KEYS.FEATURE_SET, flag_name)}
-
-
-@app.get(ENDPOINTS.INFO)
-def get_info(request: Request):
-    """Get some generic information about the request, mostly for debugging"""
-    return {
-        "message": "Welcome to daq-config API.",
-        "root_path": request.scope.get("root_path"),
-        "request_headers": request.headers,
-    }
-
-
-if DEV_MODE:
-
-    @app.api_route("/{full_path:path}")
-    async def catch_all(request: Request, full_path: str):
-        return {
-            "message": "resource not found, supplying info for debug",
-            "root_path": request.scope.get("root_path"),
-            "path": full_path,
-            "request_headers": repr(request.headers),
-        }
-
-
-def _load_beamline_params():
-    global BEAMLINE_PARAMS
-    BEAMLINE_PARAMS = GDABeamlineParameters.from_file(BEAMLINE_PARAM_PATH)
 
+@app.get(ENDPOINTS.CONFIG + "/{file_path:path}")
+def get_configuration(file_path: Path):
+    """Read a file and return its contents completely unformatted as a string. After
+    https://github.com/DiamondLightSource/daq-config-server/issues/67, this endpoint
+    will convert commonly read files to a dictionary format
+    """
+    if not file_path.exists():
+        raise FileNotFoundError(f"File {file_path} cannot be found")
 
-def _set_beamline_param_path(dev: bool = True):
-    global BEAMLINE_PARAM_PATH
-    if dev:
-        BEAMLINE_PARAM_PATH = "tests/test_data/beamline_parameters.txt"
-    else:
-        BEAMLINE_PARAM_PATH = BEAMLINE_PARAMETER_PATHS["i03"]
+    with file_path.open("r", encoding="utf-8") as f:
+        return f.read()
 
 
-def main(args):
-    _set_beamline_param_path(args.dev)
-    _load_beamline_params()
+def main():
     uvicorn.run(app="daq_config_server.app:app", host="0.0.0.0", port=8555)

src/daq_config_server/beamline_parameters.py

@@ -1,5 +1,5 @@
 from logging import Logger, getLogger
-from typing import Any, TypeVar
+from typing import TypeVar
 
 import requests
 
@@ -18,65 +18,12 @@ def _get(
         self,
         endpoint: str,
         item: str | None = None,
-        options: dict[str, str] | None = None,
-        data: dict[str, Any] | None = None,
     ):
-        r = requests.get(
-            self._url + endpoint + (f"/{item}" if item else ""), options, json=data
-        )
+        r = requests.get(self._url + endpoint + (f"/{item}" if item else ""))
         return r.json()
 
-    def get_beamline_param(self, param: str) -> BlParamDType | None:
-        return self._get(ENDPOINTS.BL_PARAM, param).get(param)
-
-    def get_some_beamline_params(self, params: list[str]) -> dict[str, BlParamDType]:
-        return self._get(ENDPOINTS.BL_PARAM, data={"param_list": params})
-
-    def get_all_beamline_params(self) -> dict[str, BlParamDType]:
-        return self._get(ENDPOINTS.BL_PARAM)
-
-    def get_feature_flag(self, flag_name: str) -> bool | None:
-        """Get the specified feature flag; returns None if it does not exist. Will check
-        that the HTTP response is correct and raise an AssertionError if not."""
-        return self._get(ENDPOINTS.FEATURE, flag_name).get(flag_name)
-
-    def get_all_feature_flags(self) -> dict | None:
-        """Get the values for all flags; returns None if it does not exist. Will check
-        that the HTTP response is correct and raise an AssertionError if not."""
-        return self._get(ENDPOINTS.FEATURE, options={"get_values": "true"})
-
-    def get_feature_flag_list(self) -> list[str]:
-        """Get the list of all feature flags. Will check that the HTTP response is
-        correct and raise an AssertionError if not."""
-        return self._get(ENDPOINTS.FEATURE)
-
-    def best_effort_get_feature_flag(
-        self, flag_name: str, fallback: T = None
-    ) -> bool | T:
-        """Get the specified feature flag, returns fallback value (default None) if it
-        doesn't exist or if there is a connection error - in the latter case logs to
-        error."""
-        try:
-            assert (
-                result := self._get(ENDPOINTS.FEATURE, flag_name).get(flag_name)
-            ) is not None
-            return result
-        except (AssertionError, OSError):
-            self._log.error(
-                "Encountered an error reading from the config service.", exc_info=True
-            )
-            return fallback
-
-    def best_effort_get_all_feature_flags(self) -> dict[str, bool]:
-        """Get all flags, returns an empty dict if there are no flags, or if there
-        is a connection error - in the latter case logs to error."""
-        try:
-            assert (
-                result := self._get(ENDPOINTS.FEATURE, options={"get_values": "true"})
-            ) is not None
-            return result
-        except (AssertionError, OSError):
-            self._log.error(
-                "Encountered an error reading from the config service.", exc_info=True
-            )
-            return {}
+    def read_unformatted_file(self, file_path: str) -> BlParamDType | None:
+        # After https://github.com/DiamondLightSource/daq-config-server/issues/67, we
+        # can get specific formats, and then have better typing on
+        # return values
+        return self._get(ENDPOINTS.CONFIG, file_path)

src/daq_config_server/client.py

@@ -2,17 +2,5 @@
 
 
 @dataclass(frozen=True)
-class Endpoints:
-    FEATURE = "/featureflag"
-    FEATURE_DISALLOWED_SPECIALS = ["all"]
-    BL_PARAM = "/beamlineparameters"
-    INFO = "/info"
-
-
-@dataclass(frozen=True)
-class DatabaseKeys:
-    FEATURE_SET = "featureflags"
-
-
-DATABASE_KEYS = DatabaseKeys()
-ENDPOINTS = Endpoints()
+class ENDPOINTS:
+    CONFIG = "/config"