Add presence lookup to Python SDK

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 0.1.1 - 2026-07-17
+
+- Added `get_presence(channel)` for authoritative occupancy and member lookups.
+
 ## 0.1.0 - 2026-06-12
 
 Initial public release.

README.md

@@ -199,6 +199,7 @@ python examples/publish_p5.py
 | `connect_sync(timeout=10)` | Same, but runs in a background thread |
 | `await disconnect()` | Close connection and stop run loop |
 | `disconnect_sync(timeout=10)` | Close a sync/background-thread connection |
+| `await get_presence(channel)` | Return authoritative `{occupancy, members}` presence data |
 | `subscribe(channel, handler)` | Register an async message handler |
 | `unsubscribe(channel, handler=None)` | Remove handler (or all) from channel |
 | `await publish(channel, data, content_type=None, metadata=None)` | Send JSON, or auto-detect bytes-like binary data |
@@ -209,6 +210,14 @@ python examples/publish_p5.py
 | `subscribe_any(channel, handler)` | Register an async handler for JSON and binary messages |
 | `on(event, handler)` | Register an event handler (decorator or direct) |
 
+`get_presence()` reuses the JWT from `connect()`, requires the API key's
+`presence` scope, and should be called occasionally or on a throttled timer:
+
+```python
+presence = await dn.get_presence("project.abc.demo")
+print(presence["occupancy"], presence["members"])
+```
+
 ### Events
 
 | Event | Handler signature | Fired when |

datanet/__init__.py

@@ -13,6 +13,7 @@
     DataNet,
     DataNetError,
     MessageMeta,
+    PresenceResult,
     base64_to_binary,
     binary_to_base64,
     build_art_dmx_packet,
@@ -25,9 +26,10 @@
     "DataNet",
     "DataNetError",
     "MessageMeta",
+    "PresenceResult",
     "base64_to_binary",
     "binary_to_base64",
     "build_art_dmx_packet",
     "build_dmx_frame",
 ]
-__version__ = "0.1.0"
+__version__ = "0.1.1"

datanet/client.py

@@ -38,7 +38,7 @@
 import time
 from collections import defaultdict
 from dataclasses import dataclass
-from typing import Any, Callable, Coroutine
+from typing import Any, Callable, Coroutine, TypedDict
 
 import aiohttp
 import websockets
@@ -51,6 +51,7 @@
     "DataNet",
     "DataNetError",
     "MessageMeta",
+    "PresenceResult",
     "base64_to_binary",
     "binary_to_base64",
     "build_art_dmx_packet",
@@ -107,6 +108,13 @@ class AnyMessage:
     meta: MessageMeta | BinaryMessageMeta
 
 
+class PresenceResult(TypedDict):
+    """Authoritative server-side occupancy for a channel."""
+
+    occupancy: int
+    members: list[str]
+
+
 class DataNetError(RuntimeError):
     """Structured error returned by the DataNet API or gateway."""
 
@@ -380,6 +388,76 @@ async def disconnect(self) -> None:
             self._ws = None
         await self._emit("disconnect")
 
+    async def get_presence(self, channel: str) -> PresenceResult:
+        """Return authoritative occupancy and member IDs for *channel*.
+
+        The client must be connected first so this request can reuse its
+        short-lived JWT. The API key must include the ``presence`` scope.
+        """
+        if not self._jwt:
+            raise DataNetError(
+                "DataNet: connect before requesting presence",
+                code="not_connected",
+            )
+
+        params = {"channel": channel}
+        project_id = self._jwt_project_id()
+        if project_id:
+            params["projectId"] = project_id
+
+        url = f"{self._api_url}/presence"
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(
+                    url,
+                    params=params,
+                    headers={"Authorization": f"Bearer {self._jwt}"},
+                    timeout=aiohttp.ClientTimeout(total=10),
+                ) as resp:
+                    if resp.status != 200:
+                        text = await resp.text()
+                        detail = text
+                        try:
+                            body = json.loads(text)
+                            if isinstance(body, dict):
+                                detail = str(body.get("error") or text)
+                        except json.JSONDecodeError:
+                            pass
+                        raise DataNetError(
+                            f"DataNet presence failed ({resp.status}): {detail}",
+                            code="presence_forbidden" if resp.status == 403 else "presence_failed",
+                            channel=channel,
+                            status=resp.status,
+                        )
+                    body = await resp.json()
+        except DataNetError:
+            raise
+        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
+            raise DataNetError(
+                "DataNet: presence request failed",
+                code="presence_failed",
+                channel=channel,
+            ) from exc
+
+        occupancy_value = body.get("occupancy", body.get("count", 0))
+        occupancy = occupancy_value if isinstance(occupancy_value, int) else 0
+        raw_members = body.get("members", [])
+        members = [member for member in raw_members if isinstance(member, str)] if isinstance(raw_members, list) else []
+        return {"occupancy": occupancy, "members": members}
+
+    def _jwt_project_id(self) -> str | None:
+        """Decode the unverified ``pid`` claim from the current gateway JWT."""
+        if not self._jwt:
+            return None
+        try:
+            segment = self._jwt.split(".")[1]
+            padded = segment + "=" * (-len(segment) % 4)
+            payload = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8"))
+            project_id = payload.get("pid")
+            return project_id if isinstance(project_id, str) and project_id else None
+        except (IndexError, ValueError, UnicodeDecodeError, json.JSONDecodeError):
+            return None
+
     def subscribe(self, channel: str, handler: Handler) -> None:
         """Register *handler* to be called when a message arrives on *channel*.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "datanet-sdk"
-version = "0.1.0"
+version = "0.1.1"
 description = "DataNet Python SDK — async realtime pub/sub client for the DataNet platform"
 readme = "README.md"
 license = { text = "MIT" }

tests/test_client.py

@@ -1,3 +1,4 @@
+import base64
 import json
 import unittest
 from unittest.mock import AsyncMock, patch
@@ -62,6 +63,36 @@ def post(self, url, **kwargs):
         return FakePostResponse()
 
 
+class FakePresenceResponse:
+    status = 200
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        return None
+
+    async def json(self):
+        return {"occupancy": 2, "members": ["one", "two"]}
+
+    async def text(self):
+        return ""
+
+
+class FakePresenceSession:
+    calls = []
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        return None
+
+    def get(self, url, **kwargs):
+        self.calls.append((url, kwargs))
+        return FakePresenceResponse()
+
+
 class DataNetClientTests(unittest.IsolatedAsyncioTestCase):
     async def test_pub_messages_dispatch_to_matching_handlers(self):
         client = DataNet("ak_test")
@@ -118,6 +149,27 @@ def test_disconnect_sync_without_active_connection_is_safe(self):
 
         self.assertFalse(client.connected)
 
+    async def test_get_presence_uses_current_jwt(self):
+        encoded = base64.urlsafe_b64encode(json.dumps({"pid": "project-id"}).encode()).decode().rstrip("=")
+        client = DataNet("ak_test", api_url="https://api.example.test")
+        client._jwt = f"header.{encoded}.signature"
+        FakePresenceSession.calls = []
+
+        with patch.object(client_module.aiohttp, "ClientSession", FakePresenceSession):
+            result = await client.get_presence("project.project-id.demo")
+
+        self.assertEqual(result, {"occupancy": 2, "members": ["one", "two"]})
+        url, kwargs = FakePresenceSession.calls[0]
+        self.assertEqual(url, "https://api.example.test/presence")
+        self.assertEqual(kwargs["params"], {"channel": "project.project-id.demo", "projectId": "project-id"})
+        self.assertEqual(kwargs["headers"]["Authorization"], f"Bearer {client._jwt}")
+
+    async def test_get_presence_requires_connection(self):
+        client = DataNet("ak_test")
+
+        with self.assertRaisesRegex(RuntimeError, "connect before"):
+            await client.get_presence("project.demo.room")
+
     async def test_binary_messages_dispatch_to_binary_and_any_handlers(self):
         client = DataNet("ak_test")
         binary_seen = []