Support pinned episode versions in streaming and topic APIs

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,
             },

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
+        )

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,
     )