add graphql endpoints to match the REST logic

pyproject.toml

@@ -17,7 +17,7 @@ readme = "README.md"
 requires-python = ">=3.11"
 
 [project.optional-dependencies]
-server = ["fastapi", "uvicorn", "redis", "hiredis"]
+server = ["fastapi", "uvicorn", "redis", "hiredis", "strawberry-graphql"]
 dev = [
     "copier",
     "httpx",

src/daq_config_server/app.py

@@ -1,19 +1,122 @@
 from os import environ
 
+import strawberry
 import uvicorn
 from fastapi import FastAPI, Request, Response, status
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 from redis import Redis
+from strawberry.fastapi import GraphQLRouter
 
 from .beamline_parameters import (
     BEAMLINE_PARAMETER_PATHS,
     GDABeamlineParameters,
 )
 from .constants import DATABASE_KEYS, ENDPOINTS
 
+BEAMLINE_PARAM_PATH = ""
+BEAMLINE_PARAMS: GDABeamlineParameters | None = None
 DEV_MODE = bool(int(environ.get("DEV_MODE") or 0))
 
+
+@strawberry.type
+class BeamlineParameterModel(BaseModel):
+    key: str
+    value: str | None
+
+
+@strawberry.experimental.pydantic.pydantic_model(
+    model=BeamlineParameterModel, all_fields=True
+)
+class BeamlineParameter:
+    pass
+
+
+class FeatureFlagModel(BaseModel):
+    name: str
+    value: bool
+
+
+@strawberry.experimental.pydantic.pydantic_model(
+    model=FeatureFlagModel, all_fields=True
+)
+class FeatureFlag:
+    pass
+
+
+@strawberry.type
+class Query:
+    @strawberry.field
+    def beamline_parameter(self, key: str) -> BeamlineParameter | None:
+        """Fetch a single beamline parameter"""
+        if BEAMLINE_PARAMS is None:
+            return None
+
+        value = BEAMLINE_PARAMS.params.get(key)
+        if value is None:
+            return None
+        return BeamlineParameter(key=key, value=value)
+
+    @strawberry.field
+    def all_beamline_parameters(
+        self, keys: list[str] | None = None
+    ) -> list[BeamlineParameter]:
+        """Fetch multiple beamline parameters (all or filtered)"""
+        if BEAMLINE_PARAMS is None:
+            return []
+        if keys is None:
+            return [
+                BeamlineParameter(key=k, value=v)
+                for k, v in BEAMLINE_PARAMS.params.items()
+            ]
+        return [
+            BeamlineParameter(key=k, value=BEAMLINE_PARAMS.params.get(k))
+            for k in keys
+            if k in BEAMLINE_PARAMS.params
+        ]
+
+    @strawberry.field
+    def feature_flags(self, get_values: bool = False) -> list[FeatureFlag]:
+        """Get all feature flags (as names or with values)"""
+        flags = valkey.smembers("feature_flags")  # Set name in Redis
+        if not get_values:
+            return [
+                FeatureFlag(name=flag, value=False) for flag in flags
+            ]  # Only names, values ignored
+        return [
+            FeatureFlag(name=flag, value=bool(int(valkey.get(flag)))) for flag in flags
+        ]
+
+
+@strawberry.type
+class Mutation:
+    @strawberry.mutation
+    def create_feature_flag(self, name: str, value: bool = False) -> FeatureFlag:
+        """Create a new feature flag. If it already exists, return an error."""
+        if valkey.sismember("feature_flags", name):
+            raise ValueError(f"Feature flag '{name}' already exists!")
+        valkey.sadd("feature_flags", name)
+        valkey.set(name, int(value))
+        return FeatureFlag(name=name, value=value)
+
+    @strawberry.mutation
+    def update_feature_flag(self, name: str, value: bool) -> FeatureFlag:
+        """Update the value of an existing feature flag."""
+        if not valkey.sismember("feature_flags", name):
+            raise ValueError(f"Feature flag '{name}' does not exist!")
+        valkey.set(name, int(value))
+        return FeatureFlag(name=name, value=value)
+
+    @strawberry.mutation
+    def delete_feature_flag(self, name: str) -> bool:
+        """Delete a feature flag. Returns True if deleted, False if not found."""
+        if not valkey.sismember("feature_flags", name):
+            return False
+        valkey.srem("feature_flags", name)
+        valkey.delete(name)
+        return True
+
+
 ROOT_PATH = "/api"
 print(f"{DEV_MODE=}")
 print(f"{ROOT_PATH=}")
@@ -36,12 +139,13 @@
     allow_headers=["*"],
 )
 
-valkey = Redis(host="localhost", port=6379, decode_responses=True)
+graphql_schema = strawberry.Schema(query=Query, mutation=Mutation)
 
-__all__ = ["main"]
+graphql_app = GraphQLRouter(graphql_schema)
+app.include_router(graphql_app, prefix="/graphql")
 
-BEAMLINE_PARAM_PATH = ""
-BEAMLINE_PARAMS: GDABeamlineParameters | None = None
+valkey = Redis(host="localhost", port=6379, decode_responses=True)
+__all__ = ["main"]
 
 
 @app.get(ENDPOINTS.BL_PARAM + "/{param}")

tests/test_graphql.py

@@ -0,0 +1,129 @@
+from fastapi.testclient import TestClient
+
+from daq_config_server.app import app
+
+client = TestClient(app)
+
+GRAPHQL_ENDPOINT = "/graphql"
+
+
+def graphql_query(query: str, variables: dict | None = None):
+    """Helper function to send GraphQL queries."""
+    if variables is None:
+        variables = {}
+    response = client.post(
+        GRAPHQL_ENDPOINT, json={"query": query, "variables": variables}
+    )
+    return response.json()
+
+
+def test_fetch_single_beamline_parameter():
+    query = """
+    query ($key: String!) {
+        beamlineParameter(key: $key) {
+            key
+            value
+        }
+    }
+    """
+    variables = {"key": "energy"}
+    response = graphql_query(query, variables)
+
+    assert "errors" not in response
+    assert response["data"]["beamlineParameter"]["key"] == "energy"
+    assert response["data"]["beamlineParameter"]["value"] is not None
+
+
+def test_fetch_all_beamline_parameters():
+    query = """
+    query {
+        allBeamlineParameters {
+            key
+            value
+        }
+    }
+    """
+    response = graphql_query(query)
+
+    assert "errors" not in response
+    assert isinstance(response["data"]["allBeamlineParameters"], list)
+    assert len(response["data"]["allBeamlineParameters"]) > 0
+
+
+def test_fetch_feature_flags():
+    query = """
+    query {
+        featureFlags {
+            name
+        }
+    }
+    """
+    response = graphql_query(query)
+
+    assert "errors" not in response
+    assert isinstance(response["data"]["featureFlags"], list)
+
+
+def test_fetch_feature_flags_with_values():
+    query = """
+    query {
+        featureFlags(getValues: true) {
+            name
+            value
+        }
+    }
+    """
+    response = graphql_query(query)
+
+    assert "errors" not in response
+    assert isinstance(response["data"]["featureFlags"], list)
+    for flag in response["data"]["featureFlags"]:
+        assert isinstance(flag["name"], str)
+        assert isinstance(flag["value"], bool)
+
+
+def test_create_feature_flag():
+    mutation = """
+    mutation ($name: String!, $value: Boolean!) {
+        createFeatureFlag(name: $name, value: $value) {
+            name
+            value
+        }
+    }
+    """
+    variables = {"name": "dark_mode", "value": True}
+    response = graphql_query(mutation, variables)
+
+    assert "errors" not in response
+    assert response["data"]["createFeatureFlag"]["name"] == "dark_mode"
+    assert response["data"]["createFeatureFlag"]["value"] is True
+
+
+def test_update_feature_flag():
+    mutation = """
+    mutation ($name: String!, $value: Boolean!) {
+        updateFeatureFlag(name: $name, value: $value) {
+            name
+            value
+        }
+    }
+    """
+    variables = {"name": "dark_mode", "value": False}
+    response = graphql_query(mutation, variables)
+
+    assert "errors" not in response
+    assert response["data"]["updateFeatureFlag"]["name"] == "dark_mode"
+    assert response["data"]["updateFeatureFlag"]["value"] is False
+
+
+def test_delete_feature_flag():
+    mutation = """
+    mutation ($name: String!) {
+        deleteFeatureFlag(name: $name)
+    }
+    """
+    variables = {"name": "dark_mode"}
+    response = graphql_query(mutation, variables)
+
+    assert "errors" not in response
+    assert response["data"]["deleteFeatureFlag"] is True