feat(realtime): add realtime V2 (#886)

Author: grdsdev

poetry.lock

@@ -15,9 +15,9 @@ classifiers = [
 ]
 
 [tool.poetry.dependencies]
-python = "^3.8"
+python = "^3.9"
 postgrest = ">=0.14,<0.17.0"
-realtime = "^1.0.0"
+realtime = "^2.0.0"
 gotrue = ">=1.3,<3.0"
 httpx = ">=0.24,<0.28"
 storage3 = ">=0.5.3,<0.8.0"
@@ -37,7 +37,7 @@ python-dotenv = "^1.0.1"
 tests = 'poetry_scripts:run_tests'
 
 [tool.poetry.group.dev.dependencies]
-unasync-cli = "^0.0.9"
+unasync-cli = { git = "https://github.com/supabase-community/unasync-cli.git", branch = "main" }
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

pyproject.toml

@@ -16,9 +16,6 @@
 from ._sync.client import SyncStorageClient as SupabaseStorageClient
 from ._sync.client import create_client
 
-# Realtime Client
-from .lib.realtime_client import SupabaseRealtimeClient
-
 # Version
 from .version import __version__
 
@@ -31,7 +28,6 @@
     "Client",
     "SupabaseAuthClient",
     "SupabaseStorageClient",
-    "SupabaseRealtimeClient",
     "PostgrestAPIError",
     "PostgrestAPIResponse",
     "StorageException",

supabase/__init__.py

@@ -1,5 +1,5 @@
 import re
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from gotrue import AsyncMemoryStorage
 from gotrue.types import AuthChangeEvent, Session
@@ -10,6 +10,7 @@
     AsyncRPCFilterRequestBuilder,
 )
 from postgrest.constants import DEFAULT_POSTGREST_CLIENT_TIMEOUT
+from realtime import AsyncRealtimeChannel, AsyncRealtimeClient, RealtimeChannelOptions
 from storage3 import AsyncStorageClient
 from storage3.constants import DEFAULT_TIMEOUT as DEFAULT_STORAGE_CLIENT_TIMEOUT
 from supafunc import AsyncFunctionsClient
@@ -80,11 +81,11 @@ def __init__(
             auth_url=self.auth_url,
             client_options=options,
         )
-        # TODO: Bring up to parity with JS client.
-        # self.realtime: SupabaseRealtimeClient = self._init_realtime_client(
-        #     realtime_url=self.realtime_url,
-        #     supabase_key=self.supabase_key,
-        # )
+        self.realtime = self._init_realtime_client(
+            realtime_url=self.realtime_url,
+            supabase_key=self.supabase_key,
+            options=options.realtime if options else None,
+        )
         self.realtime = None
         self._postgrest = None
         self._storage = None
@@ -197,41 +198,33 @@ def functions(self):
             )
         return self._functions
 
-    #     async def remove_subscription_helper(resolve):
-    #         try:
-    #             await self._close_subscription(subscription)
-    #             open_subscriptions = len(self.get_subscriptions())
-    #             if not open_subscriptions:
-    #                 error = await self.realtime.disconnect()
-    #                 if error:
-    #                     return {"error": None, "data": { open_subscriptions}}
-    #         except Exception as e:
-    #             raise e
-    #     return remove_subscription_helper(subscription)
-
-    # async def _close_subscription(self, subscription):
-    #    """Close a given subscription
-
-    #    Parameters
-    #    ----------
-    #    subscription
-    #        The name of the channel
-    #    """
-    #    if not subscription.closed:
-    #        await self._closeChannel(subscription)
-
-    # def get_subscriptions(self):
-    #     """Return all channels the client is subscribed to."""
-    #     return self.realtime.channels
-
-    # @staticmethod
-    # def _init_realtime_client(
-    #     realtime_url: str, supabase_key: str
-    # ) -> SupabaseRealtimeClient:
-    #     """Private method for creating an instance of the realtime-py client."""
-    #     return SupabaseRealtimeClient(
-    #         realtime_url, {"params": {"apikey": supabase_key}}
-    #     )
+    def channel(
+        self, topic: str, params: RealtimeChannelOptions = {}
+    ) -> AsyncRealtimeChannel:
+        """Creates a Realtime channel with Broadcast, Presence, and Postgres Changes."""
+        return self.realtime.channel(topic, params)
+
+    def get_channels(self) -> List[AsyncRealtimeChannel]:
+        """Returns all realtime channels."""
+        return self.realtime.get_channels()
+
+    async def remove_channel(self, channel: AsyncRealtimeChannel) -> None:
+        """Unsubscribes and removes Realtime channel from Realtime client."""
+        await self.realtime.remove_channel(channel)
+
+    async def remove_all_channels(self) -> None:
+        """Unsubscribes and removes all Realtime channels from Realtime client."""
+        await self.realtime.remove_all_channels()
+
+    @staticmethod
+    def _init_realtime_client(
+        realtime_url: str, supabase_key: str, options: Optional[Dict[str, Any]]
+    ) -> AsyncRealtimeClient:
+        """Private method for creating an instance of the realtime-py client."""
+        return AsyncRealtimeClient(
+            realtime_url, token=supabase_key, params=options or {}
+        )
+
     @staticmethod
     def _init_storage_client(
         storage_url: str,
@@ -303,6 +296,9 @@ def _listen_to_auth_events(
 
         self.options.headers["Authorization"] = self._create_auth_header(access_token)
 
+        # set_auth is a coroutine, how to handle this?
+        self.realtime.set_auth(access_token)
+
 
 async def create_client(
     supabase_url: str,

supabase/_async/client.py

@@ -1,5 +1,5 @@
 import re
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from gotrue import SyncMemoryStorage
 from gotrue.types import AuthChangeEvent, Session
@@ -10,6 +10,7 @@
     SyncRPCFilterRequestBuilder,
 )
 from postgrest.constants import DEFAULT_POSTGREST_CLIENT_TIMEOUT
+from realtime import RealtimeChannelOptions, SyncRealtimeChannel, SyncRealtimeClient
 from storage3 import SyncStorageClient
 from storage3.constants import DEFAULT_TIMEOUT as DEFAULT_STORAGE_CLIENT_TIMEOUT
 from supafunc import SyncFunctionsClient
@@ -80,11 +81,11 @@ def __init__(
             auth_url=self.auth_url,
             client_options=options,
         )
-        # TODO: Bring up to parity with JS client.
-        # self.realtime: SupabaseRealtimeClient = self._init_realtime_client(
-        #     realtime_url=self.realtime_url,
-        #     supabase_key=self.supabase_key,
-        # )
+        self.realtime = self._init_realtime_client(
+            realtime_url=self.realtime_url,
+            supabase_key=self.supabase_key,
+            options=options.realtime if options else None,
+        )
         self.realtime = None
         self._postgrest = None
         self._storage = None
@@ -197,41 +198,33 @@ def functions(self):
             )
         return self._functions
 
-    #     async def remove_subscription_helper(resolve):
-    #         try:
-    #             await self._close_subscription(subscription)
-    #             open_subscriptions = len(self.get_subscriptions())
-    #             if not open_subscriptions:
-    #                 error = await self.realtime.disconnect()
-    #                 if error:
-    #                     return {"error": None, "data": { open_subscriptions}}
-    #         except Exception as e:
-    #             raise e
-    #     return remove_subscription_helper(subscription)
-
-    # async def _close_subscription(self, subscription):
-    #    """Close a given subscription
-
-    #    Parameters
-    #    ----------
-    #    subscription
-    #        The name of the channel
-    #    """
-    #    if not subscription.closed:
-    #        await self._closeChannel(subscription)
-
-    # def get_subscriptions(self):
-    #     """Return all channels the client is subscribed to."""
-    #     return self.realtime.channels
-
-    # @staticmethod
-    # def _init_realtime_client(
-    #     realtime_url: str, supabase_key: str
-    # ) -> SupabaseRealtimeClient:
-    #     """Private method for creating an instance of the realtime-py client."""
-    #     return SupabaseRealtimeClient(
-    #         realtime_url, {"params": {"apikey": supabase_key}}
-    #     )
+    def channel(
+        self, topic: str, params: RealtimeChannelOptions = {}
+    ) -> SyncRealtimeChannel:
+        """Creates a Realtime channel with Broadcast, Presence, and Postgres Changes."""
+        return self.realtime.channel(topic, params)
+
+    def get_channels(self) -> List[SyncRealtimeChannel]:
+        """Returns all realtime channels."""
+        return self.realtime.get_channels()
+
+    def remove_channel(self, channel: SyncRealtimeChannel) -> None:
+        """Unsubscribes and removes Realtime channel from Realtime client."""
+        self.realtime.remove_channel(channel)
+
+    def remove_all_channels(self) -> None:
+        """Unsubscribes and removes all Realtime channels from Realtime client."""
+        self.realtime.remove_all_channels()
+
+    @staticmethod
+    def _init_realtime_client(
+        realtime_url: str, supabase_key: str, options: Optional[Dict[str, Any]]
+    ) -> SyncRealtimeClient:
+        """Private method for creating an instance of the realtime-py client."""
+        return SyncRealtimeClient(
+            realtime_url, token=supabase_key, params=options or {}
+        )
+
     @staticmethod
     def _init_storage_client(
         storage_url: str,
@@ -303,6 +296,9 @@ def _listen_to_auth_events(
 
         self.options.headers["Authorization"] = self._create_auth_header(access_token)
 
+        # set_auth is a coroutine, how to handle this?
+        self.realtime.set_auth(access_token)
+
 
 def create_client(
     supabase_url: str,

supabase/_sync/client.py

@@ -1,4 +1,3 @@
 from supabase._async import auth_client
-from supabase.lib import realtime_client
 
-__all__ = ["auth_client", "realtime_client"]
+__all__ = ["auth_client"]

supabase/lib/__init__.py

@@ -72,6 +72,9 @@ def test_updates_the_authorization_header_on_auth_events() -> None:
     assert client.options.headers.get("Authorization") == f"Bearer {key}"
 
     mock_session = MagicMock(access_token="secretuserjwt")
+    realtime_mock = MagicMock()
+    client.realtime = realtime_mock
+
     client._listen_to_auth_events("SIGNED_IN", mock_session)
 
     updated_authorization = f"Bearer {mock_session.access_token}"
@@ -89,3 +92,5 @@ def test_updates_the_authorization_header_on_auth_events() -> None:
 
     assert client.storage.session.headers.get("apiKey") == key
     assert client.storage.session.headers.get("Authorization") == updated_authorization
+
+    realtime_mock.set_auth.assert_called_once_with(mock_session.access_token)

supabase/lib/realtime_client.py

@@ -0,0 +1,14 @@
+import supabase
+
+
+def test_functions_client_initialization() -> None:
+    ref = "ooqqmozurnggtljmjkii"
+    url = f"https://{ref}.supabase.co"
+    # Sample JWT Key
+    key = "xxxxxxxxxxxxxx.xxxxxxxxxxxxxxx.xxxxxxxxxxxxxxx"
+    sp = supabase.Client(url, key)
+    assert sp.realtime_url == f"wss://{ref}.supabase.co/realtime/v1"
+
+    url = "http://localhost:54322"
+    sp_local = supabase.Client(url, key)
+    assert sp_local.realtime_url == f"ws://localhost:54322/realtime/v1"