diff --git a/CHANGELOG.md b/CHANGELOG.md index 539acff..151fb6a 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index 68b2872..c67f97f 100644 --- a/README.md +++ b/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 | diff --git a/datanet/__init__.py b/datanet/__init__.py index 0c3237d..0467df5 100644 --- a/datanet/__init__.py +++ b/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" diff --git a/datanet/client.py b/datanet/client.py index 726f902..ab3ef4a 100644 --- a/datanet/client.py +++ b/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*. diff --git a/pyproject.toml b/pyproject.toml index 8098e2d..fcdae70 100644 --- a/pyproject.toml +++ b/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" } diff --git a/tests/test_client.py b/tests/test_client.py index 8333640..4c65f9e 100644 --- a/tests/test_client.py +++ b/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 = []