Add OpaClient to wrap OPA interactions

Author: tpoliaw

helm/blueapi/config_schema.json

@@ -340,6 +340,11 @@
                     "minLength": 1,
                     "title": "Root",
                     "type": "string"
+                },
+                "audience": {
+                    "default": "account",
+                    "title": "Audience",
+                    "type": "string"
                 }
             },
             "title": "OpaConfig",

helm/blueapi/values.schema.json

@@ -756,6 +756,11 @@
             "title": "OpaConfig",
             "type": "object",
             "properties": {
+                "audience": {
+                    "title": "Audience",
+                    "default": "account",
+                    "type": "string"
+                },
                 "root": {
                     "title": "Root",
                     "default": "http://localhost:8181/",

src/blueapi/config.py

@@ -298,6 +298,7 @@ class Tag(StrEnum):
 
 class OpaConfig(BlueapiBaseModel):
     root: HttpUrl = HttpUrl("http://localhost:8181")
+    audience: str = "account"
 
 
 class ApplicationConfig(BlueapiBaseModel):

src/blueapi/service/authorization.py

@@ -0,0 +1,45 @@
+import logging
+from collections.abc import Mapping
+from contextlib import AbstractAsyncContextManager, aclosing, nullcontext
+from typing import Any, Self
+
+from aiohttp import ClientSession
+
+from blueapi.config import OpaConfig
+
+LOGGER = logging.getLogger(__name__)
+
+
+class OpaClient:
+    def __init__(self, instrument: str, config: OpaConfig):
+        LOGGER.info("Creating OpaClient for %s with config %s", instrument, config)
+        self._instrument = instrument
+        self._config = config
+        self._session = ClientSession(base_url=config.root.encoded_string())
+        self._audience = config.audience
+
+    async def aclose(self):
+        LOGGER.info("Closing OPA session")
+        await self._session.close()
+
+    async def _call_opa(self, endpoint: str, data: Mapping[str, Any]) -> bool:
+        resp = await self._session.post(
+            endpoint,
+            json={
+                "input": {
+                    "beamline": self._instrument,
+                    "audience": self._audience,
+                    **data,
+                }
+            },
+        )
+        return (await resp.json())["result"]
+
+    @classmethod
+    def for_config(
+        cls, instrument: str, config: OpaConfig | None
+    ) -> AbstractAsyncContextManager[Self | None]:
+        if config:
+            return aclosing(cls(instrument, config))
+        LOGGER.info("No OPA config provided - not creating OpaClient")
+        return nullcontext()

tests/unit_tests/test_config.py

@@ -339,6 +339,7 @@ def test_config_yaml_parsed(temp_yaml_config_file):
             },
             "opa": {
                 "root": "http://opa.example.com/",
+                "audience": "account",
             },
         },
         {