Add ability to authenticate with blueapi #40

DominicOram · 2026-06-12T14:38:32Z

Should: I think it would be good to have some unit tests on this

DominicOram · 2026-06-12T14:38:29Z

Should: I think it would be good to have some unit tests on this

-Original file line number
+Diff line change
@@ Expand Up / @@ -60,6 +60,22 @@ spec: @@
               {{- with .Values.volumeMounts }}
                 {{- toYaml . | nindent 12 }}
               {{- end }}
+              {{- if or .Values.udcSecret.enabled .Values.env }}
+              env:
+                {{- if .Values.udcSecret.enabled }}
+                - name: UDC_SECRET
+                  valueFrom:
+                    secretKeyRef:
+                      name: {{ .Values.udcSecret.name }}
+                      key: {{ .Values.udcSecret.key }}
+                - name: UDC_CLIENT_ID
+                  value: "{{ .Values.udcSecret.clientId }}"
+                {{- end }}
+                {{- with .Values.env }}
+                {{- toYaml . | nindent 12 }}
+                {{- end }}
+              {{- end }}
           volumes:
           {{- with .Values.volumes }}
             {{- toYaml . | nindent 8 }}
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -75,3 +75,11 @@ volumeMounts: [] @@
     volumes: []
     nodeSelector: {}
     tolerations: []
+    env: []
+    udcSecret:
+      enabled: false
+      name: ""
+      key: udc-secret
+      clientId: ""

-Original file line number
+Diff line change
@@ Expand Up / @@ -13,7 +13,7 @@ classifiers = [ @@
     ]
     description = "A service to queue tasks and chain BlueAPI calls"
     dependencies = [
-        "blueapi>=1.13.0",
+        "blueapi>=1.14.0",
         "fastapi>=0.136.0",
         "pydantic>=2.13.2",
     ]
@@ Expand Down @@

-Original file line number
+Diff line change
@@ -1,6 +1,5 @@
     import asyncio
     import json
-    import logging
     from collections.abc import AsyncGenerator, Callable
     from blueapi.client.rest import (
@@ Expand All / @@ -16,6 +15,7 @@ @@
     from daq_queuing_service.app._config import AppConfig, load_config
     from daq_queuing_service.blueapi_interaction.blueapi_call import BlueapiCallResponse
     from daq_queuing_service.broadcaster import Broadcaster
+    from daq_queuing_service.log import LOGGER
     from daq_queuing_service.task import ExperimentDefinition, Status, Task
     from daq_queuing_service.task_queue.queue import (
         QUEUE_EVENTS,
@@ Expand All / @@ -26,8 +26,6 @@ @@
     # pyright: reportUnusedFunction=false
-    LOGGER = logging.getLogger(__name__)
     class InvalidExperimentDefinitionsError(Exception):
         def __init__(self, errors: dict[int, InvalidParametersError | UnknownPlanError]):
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -3,14 +3,13 @@ @@
     from contextlib import asynccontextmanager
     from typing import NoReturn
-    from blueapi.client import BlueapiClient
-    from blueapi.client.rest import BlueapiRestClient
     from fastapi import FastAPI
     from fastapi.middleware.cors import CORSMiddleware
     from daq_queuing_service.api.api import create_api_router
     from daq_queuing_service.api.errors import register_exception_handlers
     from daq_queuing_service.blueapi_interaction.blueapi_adapter import BlueapiClientAdapter
+    from daq_queuing_service.blueapi_interaction.clients import get_blueapi_clients
     from daq_queuing_service.broadcaster import Broadcaster
     from daq_queuing_service.plugins.construct_task_request import (
         construct_blueapi_task_request,
@@ Expand Down Expand Up / @@ -68,8 +67,7 @@ def log_task_exception(task: asyncio.Task[NoReturn]): @@
         app.state.queue = TaskQueue(converter, broadcaster)
-        blueapi_rest_client = BlueapiRestClient(config=config.blueapi.api)
-        blueapi_client = BlueapiClient.from_config(config.blueapi)
+        blueapi_rest_client, blueapi_client = get_blueapi_clients(config.blueapi)
         blueapi_client_adapter = BlueapiClientAdapter(blueapi_client)
         app.state.worker = QueueWorker(
@@ Expand Down @@

Navigation Menu

Uh oh!

Add ability to authenticate with blueapi #40

Uh oh!

Diff view

Diff view

There are no files selected for viewing

DominicOram Jun 12, 2026

Uh oh!

DominicOram Jun 12, 2026

Uh oh!

Uh oh!

Uh oh!

-Original file line number
+Diff line change
@@ -1,5 +1,4 @@
     import asyncio
-    import logging
     from dataclasses import dataclass
     from typing import Generic, TypeVar
@@ Expand All / @@ -14,7 +13,7 @@ @@
     from blueapi.service.model import TaskRequest
     from blueapi.worker import TaskStatus, WorkerState
-    LOGGER = logging.getLogger(__name__)
+    from daq_queuing_service.log import LOGGER
     T = TypeVar("T")
     E = TypeVar("E", bound=Exception)
@@ Expand Down @@

-Original file line number
+Diff line change
@@ -0,0 +1,39 @@
+    from unittest.mock import MagicMock
+    from blueapi.client import BlueapiClient
+    from blueapi.client.event_bus import EventBusClient
+    from blueapi.client.rest import BlueapiRestClient
+    from blueapi.config import ApplicationConfig
+    from bluesky_stomp.messaging import Broker, StompClient
+    from daq_queuing_service.blueapi_interaction.token_retriever import UDCTokenRetriever
+    def get_blueapi_clients(
+        blueapi_config: ApplicationConfig,
+    ) -> tuple[BlueapiRestClient, BlueapiClient]:
+        if not blueapi_config.oidc:
+            blueapi_config.oidc = MagicMock()
+        blueapi_rest_client = BlueapiRestClient(
+            config=blueapi_config.api,
+            #  Waiting on https://github.com/DiamondLightSource/blueapi/pull/1553
+            session_manager=UDCTokenRetriever(),  # type: ignore
+        )
+        if blueapi_config.stomp.enabled:
+            assert blueapi_config.stomp.url.host is not None, "Stomp URL missing host"
+            assert blueapi_config.stomp.url.port is not None, "Stomp URL missing port"
+            stomp_client = StompClient.for_broker(
+                broker=Broker(
+                    host=blueapi_config.stomp.url.host,
+                    port=blueapi_config.stomp.url.port,
+                    auth=blueapi_config.stomp.auth,
+                )
+            )
+            events = EventBusClient(stomp_client)
+            blueapi_client = BlueapiClient(blueapi_rest_client, events)
+        else:
+            blueapi_client = BlueapiClient(blueapi_rest_client)
+        return blueapi_rest_client, blueapi_client

-Original file line number
+Diff line change
@@ -0,0 +1,47 @@
+    import os
+    import requests
+    from daq_queuing_service.log import LOGGER
+    class UDCTokenRetriever:
+        """Implements `get_valid_access_token` to get a token using a sealed secret."""
+        def __init__(
+            self,
+            secret_variable_name: str = "UDC_SECRET",
+            client_id_variable_name: str = "UDC_CLIENT_ID",
+        ):
+            self._secret_variable_name = secret_variable_name
+            self._client_id_variable_name = client_id_variable_name
+        def get_valid_access_token(self) -> str:
+            token_url = (
+                "https://identity.diamond.ac.uk/realms/dls/protocol/openid-connect/token"
+            )
+            client_id = os.environ.get(self._client_id_variable_name)
+            client_secret = os.environ.get(self._secret_variable_name)
+            if not client_secret:
+                LOGGER.debug("No UDC secret found")
+                return ""
+            if not client_id:
+                LOGGER.debug("No UDC client ID found")
+                return ""
+            LOGGER.debug("Found UDC secret")
+            response = requests.post(
+                token_url,
+                data={
+                    "client_id": client_id,
+                    "client_secret": client_secret,
+                    "grant_type": "client_credentials",
+                },
+            )
+            response.raise_for_status()
+            token = response.json().get("access_token")
+            LOGGER.debug("Returning token")
+            return token

-Original file line number
+Diff line change
@@ -0,0 +1,21 @@
+    import logging
+    import colorlog
+    HANDLER = colorlog.StreamHandler()
+    HANDLER.setFormatter(
+        colorlog.ColoredFormatter(
+            "%(log_color)s%(asctime)s [%(name)s] %(levelname)s: %(message)s",
+            log_colors={
+                "DEBUG": "cyan",
+                "INFO": "green",
+                "WARNING": "yellow",
+                "ERROR": "red",
+                "CRITICAL": "bold_red",
+            },
+        )
+    )
+    LOGGER = logging.getLogger("Queue")
+    LOGGER.addHandler(HANDLER)
+    LOGGER.setLevel(logging.DEBUG)
+    LOGGER.propagate = False

-Original file line number
+Diff line change
@@ -1,5 +1,4 @@
     import asyncio
-    import logging
     from collections.abc import Callable, Sequence
     from types import TracebackType
     from typing import Any, Literal
@@ Expand All / @@ -13,6 +12,7 @@ @@
         CallStatus,
     )
     from daq_queuing_service.broadcaster import Broadcaster, Event
+    from daq_queuing_service.log import LOGGER
     from daq_queuing_service.plugins.converter_utils import Converter
     from daq_queuing_service.task import Status, Task, TaskWithPosition
     from daq_queuing_service.task_queue.queue_utils import (
@@ Expand All / @@ -24,8 +24,6 @@ @@
         TaskNotInQueueError,
     )
-    LOGGER = logging.getLogger(__name__)
     class TaskRegistry(dict[str, Task]):
         def __missing__(self, task_id: str) -> Task:
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -17,11 +17,14 @@ @@
     from daq_queuing_service.blueapi_interaction.blueapi_adapter import BlueapiClientAdapter
     from daq_queuing_service.blueapi_interaction.blueapi_call import BlueapiCall, CallStatus
+    from daq_queuing_service.log import HANDLER
     from daq_queuing_service.task import ExperimentDefinition
     from daq_queuing_service.task_queue.queue import TaskQueue
-    LOGGER = logging.getLogger(__name__)
+    LOGGER = logging.getLogger("Queue Worker")
+    LOGGER.addHandler(HANDLER)
     LOGGER.setLevel(logging.DEBUG)
+    LOGGER.propagate = False
     class QueueWorker:
@@ Expand Down Expand Up / @@ -101,7 +104,8 @@ def _on_blueapi_event(event: AnyEvent, call: BlueapiCall): @@
                         assert worker_event.task_status
                         call.blueapi_id = worker_event.task_status.task_id
                         LOGGER.info(
-                            f"Call {call} is in progress, blueapi ID: {call.blueapi_id}"
+                            f"Putting call in progress, blueapi ID: {call.blueapi_id}. "
+                            + f"Call: ({call})"
                         )
                         call.put_in_progress()
                 case ProgressEvent():
@@ Expand Down @@

-Original file line number
+Diff line change
@@ -1,16 +1,25 @@
     import pytest
     from blueapi.service.model import TaskRequest
     from blueapi.worker.event import TaskError, TaskResult
+    from pytest import MonkeyPatch
     from daq_queuing_service.blueapi_interaction.blueapi_call import BlueapiCall
     from daq_queuing_service.broadcaster import Broadcaster
+    from daq_queuing_service.log import LOGGER
     from daq_queuing_service.plugins.construct_task_request import (
         construct_blueapi_call_list,
     )
     from daq_queuing_service.task import ExperimentDefinition, Task, TaskWithPosition
     from daq_queuing_service.task_queue.queue import TaskQueue
+    @pytest.fixture(autouse=True)
+    def propagate_logs(monkeypatch: MonkeyPatch):
+        # This is turned off in prod to avoid duplicate logs
+        # but needed in tests for caplog to receive logs
+        monkeypatch.setattr(LOGGER, "propagate", True)
     @pytest.fixture
     def tasks() -> list[Task]:
         return [
@@ Expand Down @@

-Original file line number
+Diff line change
@@ -0,0 +1,46 @@
+    from unittest.mock import MagicMock, patch
+    from blueapi.config import ApplicationConfig, RestConfig, StompConfig
+    from pydantic import HttpUrl
+    from daq_queuing_service.blueapi_interaction.clients import get_blueapi_clients
+    @patch("daq_queuing_service.blueapi_interaction.clients.UDCTokenRetriever")
+    @patch("daq_queuing_service.blueapi_interaction.clients.BlueapiClient")
+    @patch("daq_queuing_service.blueapi_interaction.clients.BlueapiRestClient")
+    def test_get_blueapi_clients_constructs_clients_with_expected_args_and_returns_clients(
+        mock_rest_client: MagicMock,
+        mock_blueapi_client: MagicMock,
+        mock_token_retriever: MagicMock,
+    ):
+        rest_config = RestConfig(url=HttpUrl("http://test_url.com"))
+        rest_client, blueapi_client = get_blueapi_clients(
+            ApplicationConfig(api=rest_config)
+        )
+        mock_rest_client.assert_called_once_with(
+            config=rest_config, session_manager=mock_token_retriever.return_value
+        )
+        mock_blueapi_client.assert_called_once_with(rest_client)
+        assert rest_client is mock_rest_client.return_value
+        assert blueapi_client is mock_blueapi_client.return_value
+    @patch("daq_queuing_service.blueapi_interaction.clients.EventBusClient")
+    @patch("daq_queuing_service.blueapi_interaction.clients.BlueapiClient")
+    @patch("daq_queuing_service.blueapi_interaction.clients.BlueapiRestClient")
+    def test_get_blueapi_clients_constructs_blueapi_client_with_stomp_if_enabled_in_config(
+        mock_rest_client: MagicMock,
+        mock_blueapi_client: MagicMock,
+        mock_event_bus_client: MagicMock,
+    ):
+        rest_config = RestConfig(url=HttpUrl("http://test_url.com"))
+        rest_client, _ = get_blueapi_clients(
+            ApplicationConfig(api=rest_config, stomp=StompConfig(enabled=True))
+        )
+        mock_blueapi_client.assert_called_once_with(
+            rest_client, mock_event_bus_client.return_value
+        )

Uh oh!

Add ability to authenticate with blueapi #40

Uh oh!

Add ability to authenticate with blueapi #40

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

DominicOram Jun 12, 2026

Choose a reason for hiding this comment

Uh oh!

DominicOram Jun 12, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!