Add OpaClient to wrap OPA interactions

Author: tpoliaw

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()