From a4229f2a384d7b188dfcee4e077a733ba1f0c78c Mon Sep 17 00:00:00 2001 From: Mike Guida Date: Thu, 29 Oct 2026 15:24:49 -0600 Subject: [PATCH] Expose episode versions in streaming and topic APIs --- foxglove/client/api.py | 16 ++++++++++++++ tests/test_episodes.py | 41 +++++++++++++++++++++++++++++++++++ tests/test_stream_messages.py | 1 + 3 files changed, 58 insertions(+) diff --git a/foxglove/client/api.py b/foxglove/client/api.py index 521b1e2..471d140 100644 --- a/foxglove/client/api.py +++ b/foxglove/client/api.py @@ -450,6 +450,7 @@ def iter_messages( decoder_factories: Optional[List[DecoderFactory]] = None, project_id: Optional[str] = None, episode_id: Optional[str] = None, + episode_version: Optional[int] = None, ): """ yields a stream of (schema, channel, message, decoded message) values. @@ -468,6 +469,8 @@ def iter_messages( device_name as an identifier in multi-project organizations. episode_id: ID of an episode to download. Its time range is used when start and end are omitted. + episode_version: Episode version to read. Omit to use the latest version. + Requires episode_id. """ if topics is None: topics = [] @@ -481,6 +484,7 @@ def iter_messages( topics=topics, project_id=project_id, episode_id=episode_id, + episode_version=episode_version, ) response = requests.get(stream_link, stream=True) try: @@ -566,6 +570,8 @@ def _make_stream_link( or session_key is not None ): raise RuntimeError("episode_id cannot be combined with another identifier") + if episode_version is not None and episode_id is None: + raise RuntimeError("episode_version requires episode_id") if episode_id is None and (start is None or end is None): raise RuntimeError("start and end must be provided unless using episode_id") @@ -609,6 +615,7 @@ def download_data( callback: Optional[ProgressCallback] = None, project_id: Optional[str] = None, episode_id: Optional[str] = None, + episode_version: Optional[int] = None, ) -> bytes: """ Returns raw data bytes for a device and time range. @@ -629,6 +636,8 @@ def download_data( device_name as an identifier in multi-project organizations. episode_id: ID of an episode to download. Its time range is used when start and end are omitted. + episode_version: Episode version to read. Omit to use the latest version. + Requires episode_id. """ if topics is None: topics = [] @@ -645,6 +654,7 @@ def download_data( compression_format=compression_format, project_id=project_id, episode_id=episode_id, + episode_version=episode_version, ), callback=callback, ) @@ -1111,6 +1121,7 @@ def get_topics( session_id: Optional[str] = None, session_key: Optional[str] = None, episode_id: Optional[str] = None, + episode_version: Optional[int] = None, limit: Optional[int] = None, offset: Optional[int] = None, ): @@ -1127,6 +1138,8 @@ def get_topics( :param session_key: Key of a session to list topics from :param episode_id: ID of an episode to list topics from. Its time range is used when start and end are omitted. + :param episode_version: Episode version to read. Omit to use the latest version. + Requires episode_id. :param limit: Maximum topics in this page (server default and maximum: 2000). :param offset: Number of topics to skip to retrieve a subsequent page. """ @@ -1137,6 +1150,8 @@ def get_topics( or session_key is not None ): raise RuntimeError("episode_id cannot be combined with another identifier") + if episode_version is not None and episode_id is None: + raise RuntimeError("episode_version requires episode_id") if episode_id is None and (start is None or end is None): raise RuntimeError("start and end must be provided unless using episode_id") response = self.__session.get( @@ -1151,6 +1166,7 @@ def get_topics( "sessionId": session_id, "sessionKey": session_key, "episodeId": episode_id, + "episodeVersion": episode_version, "limit": limit, "offset": offset, }, diff --git a/tests/test_episodes.py b/tests/test_episodes.py index 6667bee..673daf4 100644 --- a/tests/test_episodes.py +++ b/tests/test_episodes.py @@ -292,3 +292,44 @@ def test_get_topics_without_episode_requires_time_range(): client.get_topics(session_id="ses_1") assert str(raised.value) == "start and end must be provided unless using episode_id" + + +@pytest.mark.parametrize("episode_version", [None, 2]) +@pytest.mark.parametrize("method", ["iter_messages", "download_data", "get_topics"]) +@responses.activate +def test_episode_version_requests(method, episode_version): + from .generate import generate_json_data + + link = "https://storage.example/episode.mcap" + if method == "get_topics": + responses.add(responses.GET, api_url("/v1/data/topics"), json=[]) + else: + responses.add(responses.POST, api_url("/v1/data/stream"), json={"link": link}) + responses.add(responses.GET, link, body=generate_json_data()) + result = getattr(Client("test"), method)( + episode_id="ep_1", + episode_version=episode_version, + start=NOW - timedelta(seconds=5), + end=NOW, + ) + if method == "iter_messages": + assert len(list(result)) == 10 + request = responses.calls[0].request + params = request.params if method == "get_topics" else json.loads(request.body) + assert params["episodeId"] == "ep_1" + assert params["start"] == (NOW - timedelta(seconds=5)).astimezone().isoformat() + assert params["end"] == NOW.astimezone().isoformat() + if episode_version is None: + assert "episodeVersion" not in params + else: + assert params["episodeVersion"] == ( + str(episode_version) if method == "get_topics" else episode_version + ) + + +@pytest.mark.parametrize("method", ["iter_messages", "download_data", "get_topics"]) +def test_episode_version_requires_episode_id(method): + with pytest.raises(RuntimeError, match="episode_version requires episode_id"): + getattr(Client("test"), method)( + device_id="device", start=NOW, end=NOW, episode_version=2 + ) diff --git a/tests/test_stream_messages.py b/tests/test_stream_messages.py index 6ce4718..6ffb341 100644 --- a/tests/test_stream_messages.py +++ b/tests/test_stream_messages.py @@ -60,4 +60,5 @@ def test_iter_messages_by_episode(_get): topics=[], project_id=None, episode_id="ep_1", + episode_version=None, )