Merge pull request #477 from permitio/rk/opal-0.7.1

Prepare opal 0.7.1

docker/Dockerfile

@@ -1,7 +1,7 @@
 # BUILD STAGE ---------------------------------------
 # split this stage to save time and reduce image size
 # ---------------------------------------------------
-FROM python:3.10 as BuildStage
+FROM python:3.10-bullseye as BuildStage
 # from now on, work in the /app directory
 WORKDIR /app/
 # Layer dependency install (for caching)
@@ -24,7 +24,7 @@ RUN cd /tmp/cedar-agent && \
 
 # COMMON IMAGE --------------------------------------
 # ---------------------------------------------------
-FROM python:3.10-slim as common
+FROM python:3.10-slim-bullseye as common
 
 # copy libraries from build stage (This won't copy redundant libraries we used in BuildStage)
 COPY --from=BuildStage /usr/local /usr/local

packages/__packaging__.py

@@ -8,7 +8,7 @@
 """
 import os
 
-VERSION = (0, 7, 0)
+VERSION = (0, 7, 1)
 VERSION_STRING = ".".join(map(str, VERSION))
 
 __version__ = VERSION_STRING

packages/opal-common/opal_common/topics/publisher.py

@@ -124,8 +124,8 @@ def __init__(self, endpoint: PubSubEndpoint):
         self._endpoint = endpoint
         super().__init__()
 
-    def publish(self, topics: TopicList, data: Any = None):
-        self._add_task(
+    async def publish(self, topics: TopicList, data: Any = None):
+        await self._add_task(
             asyncio.create_task(self._endpoint.publish(topics=topics, data=data))
         )
 
@@ -168,14 +168,16 @@ async def wait_until_done(self):
         context."""
         return await self._client.wait_until_done()
 
-    def publish(self, topics: TopicList, data: Any = None):
+    async def publish(self, topics: TopicList, data: Any = None):
         """publish a message by launching a background task on the event loop.
 
         Args:
             topics (TopicList): a list of topics to publish the message to
             data (Any): optional data to publish as part of the message
         """
-        self._add_task(asyncio.create_task(self._publish(topics=topics, data=data)))
+        await self._add_task(
+            asyncio.create_task(self._publish(topics=topics, data=data))
+        )
 
     async def _publish(self, topics: TopicList, data: Any = None) -> bool:
         """Do not trigger directly, must be triggered via publish() in order to
@@ -190,7 +192,7 @@ def __init__(self, endpoint: PubSubEndpoint, scope_id: str):
         super().__init__(endpoint)
         self._scope_id = scope_id
 
-    def publish(self, topics: TopicList, data: Any = None):
+    async def publish(self, topics: TopicList, data: Any = None):
         scoped_topics = [f"{self._scope_id}:{topic}" for topic in topics]
         logger.info("Publishing to topics: {topics}", topics=scoped_topics)
-        super().publish(scoped_topics, data)
+        await super().publish(scoped_topics, data)

packages/opal-server/opal_server/data/data_update_publisher.py

@@ -109,7 +109,7 @@ async def publish_data_updates(self, update: DataUpdate):
         )
 
         async with self._publisher:
-            self._publisher.publish(list(all_topic_combos), update)
+            await self._publisher.publish(list(all_topic_combos), update)
 
     async def _periodic_update_callback(
         self, update: DataSourceEntryWithPollingInterval

packages/opal-server/opal_server/scopes/service.py

@@ -94,7 +94,7 @@ async def trigger_notification(self, notification: PolicyUpdateMessageNotificati
         async with ScopedServerSideTopicPublisher(
             self._pubsub_endpoint, self._scope_id
         ) as publisher:
-            publisher.publish(notification.topics, notification.update)
+            await publisher.publish(notification.topics, notification.update)
 
 
 class ScopesService:

packages/requires.txt

@@ -2,7 +2,7 @@ charset-normalizer>=2.0.12,<3
 idna>=3.3,<4
 typer>=0.4.1,<1
 fastapi>=0.78.0,<1
-fastapi_websocket_pubsub==0.3.3
+fastapi_websocket_pubsub==0.3.4
 fastapi_websocket_rpc>=0.1.21,<1
 gunicorn>=20.1.0,<21
 pydantic[email]>=1.9.1,<2