From 9b4f340b77ca89ed5a3c4eeeae97cd16dc4817b3 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Sun, 23 Aug 2026 14:31:53 -0700 Subject: [PATCH 1/2] python(feat): expose include_received_at in get_data --- python/CHANGELOG.md | 14 + python/docs/guides/pytest_plugin/index.md | 10 +- .../_internal/low_level_wrappers/data.py | 59 +++- .../_internal/low_level_wrappers/test_data.py | 316 ++++++++++++++++++ .../_tests/resources/test_channels.py | 33 ++ .../_tests/sift_types/test_channel.py | 16 + python/lib/sift_client/resources/channels.py | 8 + .../resources/sync_stubs/__init__.pyi | 6 + python/lib/sift_client/sift_types/channel.py | 7 + 9 files changed, 459 insertions(+), 10 deletions(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index b568ef513..ee4495977 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -5,6 +5,20 @@ This project adheres to [Semantic Versioning](http://semver.org/). ## [Unreleased] +### What's New + +#### `include_received_at` on `get_data` + +`client.channels.get_data(...)`, `get_data_as_arrow(...)`, and `Channel.data(...)` now take `include_received_at`. When set, each returned DataFrame gains a `".sift_received_at"` column with the time each point was received by Sift, alongside the measurement timestamps. Not available for enum or bitfield channels. Received-at data is never cached, so these calls always fetch from the server. + +```python +data = client.channels.get_data( + channels=channels, + run=run, + include_received_at=True, +) +``` + ## [v0.20.0] - August 25, 2026 ### What's New diff --git a/python/docs/guides/pytest_plugin/index.md b/python/docs/guides/pytest_plugin/index.md index 215dc9048..6d33f1ff3 100644 --- a/python/docs/guides/pytest_plugin/index.md +++ b/python/docs/guides/pytest_plugin/index.md @@ -100,11 +100,11 @@ The plugin runs in one of three modes, picked at invocation. | **Offline** | `--sift-offline` | No; records to a log file for later replay | Environments without Sift access. | | **Disabled** | `--sift-disabled` | No | Local dev. Bounds still evaluate and return a real pass/fail. | -Online mode pings Sift once at session start and aborts if Sift is unreachable or the credentials are invalid, -so a misconfigured job fails immediately instead of silently producing no report. -During the run, every create and update is appended to a JSONL log file. -A background worker uploads new entries to Sift incrementally. -If the connection drops mid-test, the test keeps running and the log keeps writing locally. +Online mode pings Sift once at session start and aborts if Sift is unreachable or the credentials are invalid, +so a misconfigured job fails immediately instead of silently producing no report. +During the run, every create and update is appended to a JSONL log file. +A background worker uploads new entries to Sift incrementally. +If the connection drops mid-test, the test keeps running and the log keeps writing locally. The remaining entries can be uploaded afterward by running import-test-result-log, which the plugin prints on exit. That command resumes into the report the interrupted run created rather than starting a second one. See [Running Modes](running_modes.md) for the log-file and replay pipeline, diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data.py b/python/lib/sift_client/_internal/low_level_wrappers/data.py index 5a2dd522f..019f175a5 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/data.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/data.py @@ -37,6 +37,11 @@ # has been resolved. In the mean time each channel gets its own request. REQUEST_BATCH_SIZE = 1 +# Label the server attaches to the per-point ingest timestamps returned in the +# `extras` field when `GetDataRequest.include_received_at` is set. Also used as +# the suffix of the resulting ".sift_received_at" DataFrame column. +RECEIVED_AT_LABEL = "sift_received_at" + TimeRange = Tuple[datetime, datetime] @@ -571,6 +576,7 @@ async def _get_data_impl( page_size: int | None = None, page_token: str | None = None, order_by: str | None = None, + include_received_at: bool = False, ) -> tuple[list[Any], str | None]: """Get the data for a channel during a run.""" queries = [ @@ -584,6 +590,7 @@ async def _get_data_impl( "end_time": to_timestamp_pb(end_time), "page_size": page_size, "page_token": page_token, + "include_received_at": True if include_received_at else None, } request = GetDataRequest(**request_kwargs) @@ -773,6 +780,7 @@ async def get_channel_data( page_size: int | None = None, ignore_cache: bool = False, show_progress: bool = False, + include_received_at: bool = False, ) -> dict[str, pd.DataFrame]: """Get the data for a channel during a run.""" ret_data: dict[str, pd.DataFrame] = {} @@ -780,6 +788,12 @@ async def get_channel_data( start_time = start_time or datetime.fromtimestamp(0, tz=timezone.utc) end_time = end_time or datetime.now(timezone.utc) + # Received-at columns are never cached: cached segments were fetched + # without them, so mixing cached and fresh frames would yield rows + # with and without the column. Bypass both cache read and write. + if include_received_at: + ignore_cache = True + self._update_name_id_map(channels) # Two work queues. Fully uncached channels share the full range @@ -843,6 +857,7 @@ async def get_channel_data( "run_id": run_id, "start_time": start_time, "end_time": end_time, + "include_received_at": include_received_at, }, ", ".join(id_to_name.get(cid, cid) for cid in batch), ) @@ -858,6 +873,7 @@ async def get_channel_data( "run_id": run_id, "start_time": gap_start, "end_time": gap_end, + "include_received_at": include_received_at, }, f"{id_to_name.get(cid, cid)} [{gap_start:%H:%M:%S}-{gap_end:%H:%M:%S}]", ) @@ -987,9 +1003,11 @@ def _merge_pages( stitched inline via :meth:`ChannelDataCache.get_range` before dispatching wire fetches for the gaps. Cached entries are folded in as the first frame for their channel so they - participate in the same final concat; ``groupby(level=0).last()`` - preserves the previous behavior of letting a later-positioned - (fresher) value win on duplicate timestamps. + participate in the same final concat; row-level dedup keeps the + later-positioned (fresher) row whole on duplicate timestamps. + Dedup is per-row, not ``groupby.last()``: last() skips nulls per + column, which would pair a fresh value with a stale row's + received-at when the fresh point's received-at is NaT. """ per_channel_frames: dict[str, list[pd.DataFrame]] = {} for page in pages: @@ -1001,12 +1019,13 @@ def _merge_pages( for name, frames in per_channel_frames.items(): if name in ret_data: # Cached slice goes first so fresher pages (positioned later - # in the list) win on overlapping timestamps after groupby. + # in the list) win on overlapping timestamps after dedup. frames.insert(0, ret_data[name]) if len(frames) == 1: ret_data[name] = frames[0] else: - ret_data[name] = pd.concat(frames).groupby(level=0).last() + combined = pd.concat(frames) + ret_data[name] = combined[~combined.index.duplicated(keep="last")].sort_index() return ret_data @staticmethod @@ -1021,6 +1040,8 @@ def try_deserialize_channel_data(channel_data: Any) -> dict[str, pd.DataFrame]: metadata = proto_data_value.metadata ret_data = {} + received_at = DataLowLevelClient._received_at_values(proto_data_value) + components = ( proto_data_value.values if proto_data_class is BitFieldValues else [proto_data_value] ) @@ -1034,6 +1055,34 @@ def try_deserialize_channel_data(channel_data: Any) -> dict[str, pd.DataFrame]: time_column.append(to_timestamp_nanos(value_obj.timestamp)) value_column.append(value_obj.value) df = pd.DataFrame({name: value_column}, index=time_column) + if received_at is not None and len(received_at) == len(value_column): + # Explicit dtype: an all-NaT page would otherwise infer + # tz-naive and degrade the tz-aware concat in _merge_pages + # to an object column. + df[f"{name}.{RECEIVED_AT_LABEL}"] = pd.Series( + received_at, index=df.index, dtype="datetime64[ns, UTC]" + ) ret_data[name] = df return ret_data + + @staticmethod + def _received_at_values(proto_data_value: Any) -> list[pd.Timestamp] | None: + """Extract per-point ``sift_received_at`` timestamps from ``extras``. + + The server aligns one timestamp per data point when + ``GetDataRequest.include_received_at`` is set; points without one + come back as ``NaT``. Returns ``None`` when the message carries no + received-at dimension (flag unset, or a type without extras such as + enums and bitfields). + """ + for extra in getattr(proto_data_value, "extras", ()): + if extra.label != RECEIVED_AT_LABEL: + continue + if extra.WhichOneof("value_wrapper") != "dimension_proto_timestamp_values": + continue + return [ + to_timestamp_nanos(v.value) if v.HasField("value") else cast("pd.Timestamp", pd.NaT) + for v in extra.dimension_proto_timestamp_values.values + ] + return None diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py index 31fb8cd7b..57b6cac17 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data.py @@ -1858,3 +1858,319 @@ async def test_progress_text_all_cached_shows_no_rate(self, tmp_path) -> None: assert "/s" not in final, texts # nothing fetched, so no rate finally: client.channel_cache.store.close() + + +class TestReceivedAt: + """The ``include_received_at`` path: request flag, extras parsing into + a ``.sift_received_at`` column, and the cache bypass. + """ + + @staticmethod + def _double_values_any( + *, + name: str = "c1", + values: list[float], + received_at: list[datetime | None] | None = None, + start: datetime = _NOW, + ) -> Any: + """Fake wire payload: a DoubleValues proto behind a duck-typed Any. + + ``received_at`` mirrors the server contract: a ``sift_received_at`` + extras wrapper with one optional timestamp per point (``None`` for a + point without one). ``None`` for the whole arg omits extras entirely. + """ + from types import SimpleNamespace + + from sift.data.v2.data_pb2 import ( + DimensionIndividualWrapper, + DimensionProtoTimestampValues, + DoubleValue, + DoubleValues, + Metadata, + ) + + from sift_client._internal.time import to_timestamp_pb + + msg = DoubleValues( + metadata=Metadata(channel=Metadata.Channel(channel_id="cid1", name=name)), + values=[ + DoubleValue(timestamp=to_timestamp_pb(start + timedelta(milliseconds=i)), value=v) + for i, v in enumerate(values) + ], + ) + if received_at is not None: + msg.extras.append( + DimensionIndividualWrapper( + label="sift_received_at", + type=DimensionIndividualWrapper.TYPE_ADDITIONAL_TIMESTAMP_NANOS, + dimension_proto_timestamp_values=DimensionProtoTimestampValues( + values=[ + DimensionProtoTimestampValues.DimensionProtoTimestampValue( + value=to_timestamp_pb(ts) if ts is not None else None + ) + for ts in received_at + ] + ), + ) + ) + return SimpleNamespace(type_url="sift.data.v2.DoubleValues", value=msg.SerializeToString()) + + def test_deserialize_adds_prefixed_received_at_column(self) -> None: + """Extras timestamps land in a ``.sift_received_at`` column.""" + ingest = _NOW + timedelta(minutes=2) + payload = self._double_values_any(values=[1.0, 2.0], received_at=[ingest, ingest]) + + result = DataLowLevelClient.try_deserialize_channel_data(payload) + + df = result["c1"] + assert list(df.columns) == ["c1", "c1.sift_received_at"] + assert list(df["c1"]) == [1.0, 2.0] + assert all(ts == pd.Timestamp(ingest) for ts in df["c1.sift_received_at"]) + + def test_deserialize_without_extras_keeps_single_column(self) -> None: + payload = self._double_values_any(values=[1.0, 2.0]) + + result = DataLowLevelClient.try_deserialize_channel_data(payload) + + assert list(result["c1"].columns) == ["c1"] + + def test_deserialize_point_without_received_at_is_nat(self) -> None: + """A point the server has no ingest time for surfaces as ``NaT``.""" + ingest = _NOW + timedelta(minutes=2) + payload = self._double_values_any(values=[1.0, 2.0], received_at=[ingest, None]) + + result = DataLowLevelClient.try_deserialize_channel_data(payload) + + col = result["c1"]["c1.sift_received_at"] + assert col.iloc[0] == pd.Timestamp(ingest) + assert pd.isna(col.iloc[1]) + + def test_deserialize_ignores_foreign_extras(self) -> None: + """Extras wrappers that aren't the received-at dimension are skipped.""" + from sift.data.v2.data_pb2 import ( + DimensionIndividualWrapper, + DimensionStringValues, + DoubleValues, + ) + + payload = self._double_values_any(values=[1.0]) + msg = DoubleValues.FromString(payload.value) + msg.extras.append( + DimensionIndividualWrapper( + label="some_identifier", + type=DimensionIndividualWrapper.TYPE_IDENTIFIER, + dimension_string_values=DimensionStringValues( + values=[DimensionStringValues.DimensionStringValue(value="x")] + ), + ) + ) + payload.value = msg.SerializeToString() + + result = DataLowLevelClient.try_deserialize_channel_data(payload) + + assert list(result["c1"].columns) == ["c1"] + + def test_deserialize_length_mismatch_skips_column(self) -> None: + """A misaligned extras array must not produce a partial column.""" + ingest = _NOW + timedelta(minutes=2) + payload = self._double_values_any(values=[1.0, 2.0], received_at=[ingest]) + + result = DataLowLevelClient.try_deserialize_channel_data(payload) + + assert list(result["c1"].columns) == ["c1"] + + @pytest.mark.asyncio + async def test_get_data_impl_sets_flag_on_request(self) -> None: + from unittest.mock import AsyncMock + + from sift.data.v2.data_pb2 import GetDataResponse + + grpc_client = MagicMock() + client = DataLowLevelClient(grpc_client) + stub = grpc_client.get_stub.return_value + stub.GetData = AsyncMock(return_value=GetDataResponse()) + + await client._get_data_impl( + channel_ids=["c1"], end_time=_WINDOW_END, include_received_at=True + ) + + request = stub.GetData.await_args_list[0].args[0] + assert request.HasField("include_received_at") + assert request.include_received_at is True + + @pytest.mark.asyncio + async def test_get_data_impl_leaves_flag_unset_by_default(self) -> None: + from unittest.mock import AsyncMock + + from sift.data.v2.data_pb2 import GetDataResponse + + grpc_client = MagicMock() + client = DataLowLevelClient(grpc_client) + stub = grpc_client.get_stub.return_value + stub.GetData = AsyncMock(return_value=GetDataResponse()) + + await client._get_data_impl(channel_ids=["c1"], end_time=_WINDOW_END) + + request = stub.GetData.await_args_list[0].args[0] + assert not request.HasField("include_received_at") + + @pytest.mark.asyncio + async def test_include_received_at_bypasses_cache_read(self, tmp_path) -> None: + """A warm cache must not satisfy a received-at query (cached frames + carry no received-at column). + """ + client = _client_with_cache(tmp_path) + cached_df, fresh_df = _frame("c1"), _frame("c1", offset=100) + # Full-coverage segment: a plain query for this window would be a + # pure cache hit and never touch the wire. + _put(client.channel_cache, "c1", data=cached_df, seg_start=_NOW, seg_end=_WINDOW_END) + try: + with _fake_grpc(client, {"c1": [fresh_df]}) as call_log: + result = await client.get_channel_data( + channels=[_channel("c1")], + start_time=_NOW, + end_time=_WINDOW_END, + include_received_at=True, + ) + + assert call_log, "received-at query should hit the wire despite the warm cache" + for call in call_log: + assert call["include_received_at"] is True, call + pd.testing.assert_frame_equal(result["c1"].sort_index(), fresh_df.sort_index()) + finally: + client.channel_cache.store.close() + + @pytest.mark.asyncio + async def test_include_received_at_skips_cache_write(self, tmp_path) -> None: + """Received-at frames never land in the cache, so a later plain + query fetches fresh instead of serving a frame with the extra column. + """ + client = _client_with_cache(tmp_path) + try: + with _fake_grpc(client, {"c1": [_frame("c1")]}): + await client.get_channel_data( + channels=[_channel("c1")], + start_time=_NOW, + end_time=_WINDOW_END, + include_received_at=True, + ) + assert not client.channel_cache.has_any("c1") + finally: + client.channel_cache.store.close() + + @pytest.mark.asyncio + async def test_received_at_column_survives_page_merge(self) -> None: + """Two-column frames concat across pages without losing the column.""" + received = _NOW + timedelta(minutes=2) + page1, page2 = _frame("c1", rows=3), _frame("c1", rows=3, start=_NOW + timedelta(seconds=1)) + for page in (page1, page2): + page["c1.sift_received_at"] = pd.Timestamp(received) + + client = DataLowLevelClient(MagicMock()) + with _fake_grpc(client, {"c1": [page1, page2]}): + result = await client.get_channel_data( + channels=[_channel("c1")], + start_time=_NOW, + end_time=_WINDOW_END, + include_received_at=True, + ) + + df = result["c1"] + assert list(df.columns) == ["c1", "c1.sift_received_at"] + assert len(df) == 6 + assert all(ts == pd.Timestamp(received) for ts in df["c1.sift_received_at"]) + + def test_deserialize_bitfield_payload_is_unaffected(self) -> None: + """BitFieldValues has no extras field on the wire; deserialization + must not crash and must not grow received-at columns. + """ + from types import SimpleNamespace + + from sift.data.v2.data_pb2 import ( + BitFieldElementValues, + BitFieldValue, + BitFieldValues, + Metadata, + ) + + from sift_client._internal.time import to_timestamp_pb + + msg = BitFieldValues( + metadata=Metadata(channel=Metadata.Channel(channel_id="cid1", name="bf")), + values=[ + BitFieldElementValues( + name="flag_a", + values=[BitFieldValue(timestamp=to_timestamp_pb(_NOW), value=1)], + ) + ], + ) + payload = SimpleNamespace( + type_url="sift.data.v2.BitFieldValues", value=msg.SerializeToString() + ) + + result = DataLowLevelClient.try_deserialize_channel_data(payload) + + assert list(result.keys()) == ["bf.flag_a"] + assert list(result["bf.flag_a"].columns) == ["bf.flag_a"] + + def test_deserialize_enum_payload_is_unaffected(self) -> None: + """EnumValues has no extras field on the wire; deserialization must + not crash and must not grow a received-at column. + """ + from types import SimpleNamespace + + from sift.data.v2.data_pb2 import EnumValue, EnumValues, Metadata + + from sift_client._internal.time import to_timestamp_pb + + msg = EnumValues( + metadata=Metadata(channel=Metadata.Channel(channel_id="cid1", name="mode")), + values=[EnumValue(timestamp=to_timestamp_pb(_NOW), value=2)], + ) + payload = SimpleNamespace(type_url="sift.data.v2.EnumValues", value=msg.SerializeToString()) + + result = DataLowLevelClient.try_deserialize_channel_data(payload) + + assert list(result["mode"].columns) == ["mode"] + + def test_all_nat_page_keeps_tz_aware_dtype(self) -> None: + """An all-NaT page must carry the same tz-aware dtype as populated + pages so the cross-page concat doesn't degrade to object dtype. + """ + ingest = _NOW + timedelta(minutes=2) + nat_page = DataLowLevelClient.try_deserialize_channel_data( + self._double_values_any(values=[1.0, 2.0], received_at=[None, None]) + ) + populated_page = DataLowLevelClient.try_deserialize_channel_data( + self._double_values_any( + values=[3.0], received_at=[ingest], start=_NOW + timedelta(seconds=1) + ) + ) + + assert str(nat_page["c1"]["c1.sift_received_at"].dtype) == "datetime64[ns, UTC]" + + merged = DataLowLevelClient(MagicMock())._merge_pages( + [[nat_page, populated_page]], initial={} + )["c1"] + col = merged["c1.sift_received_at"] + assert str(col.dtype) == "datetime64[ns, UTC]" + assert col.isna().tolist() == [True, True, False] + assert col.iloc[-1] == pd.Timestamp(ingest) + + def test_merge_keeps_later_duplicate_row_whole(self) -> None: + """On a duplicate timestamp the fresher page's whole row wins; its + NaT received-at must not be back-filled from the stale row. + """ + ingest = _NOW + timedelta(minutes=2) + page1 = DataLowLevelClient.try_deserialize_channel_data( + self._double_values_any(values=[1.0], received_at=[ingest]) + ) + page2 = DataLowLevelClient.try_deserialize_channel_data( + self._double_values_any(values=[2.0], received_at=[None]) + ) + + merged = DataLowLevelClient(MagicMock())._merge_pages([[page1, page2]], initial={})["c1"] + + assert len(merged) == 1 + assert merged["c1"].iloc[0] == 2.0 + assert pd.isna(merged["c1.sift_received_at"].iloc[0]) diff --git a/python/lib/sift_client/_tests/resources/test_channels.py b/python/lib/sift_client/_tests/resources/test_channels.py index f337bd3f5..a2bb76e52 100644 --- a/python/lib/sift_client/_tests/resources/test_channels.py +++ b/python/lib/sift_client/_tests/resources/test_channels.py @@ -501,3 +501,36 @@ async def fake_update_channel(update): api._units_low_level_client.create_unit.assert_not_awaited() assert captured["update"].unit == "" + + +class TestGetDataForwarding: + """get_data forwards its knobs to the data low-level client. The data + path itself is covered in the low-level wrapper tests. + """ + + @pytest.mark.asyncio + async def test_forwards_include_received_at(self): + """get_data(include_received_at=True) reaches get_channel_data as True.""" + api = _make_api() + api._data_low_level_client = MagicMock() + api._data_low_level_client.get_channel_data = AsyncMock(return_value={}) + + await api.get_data( + channels=[_mock_channel(unit="")], + show_progress=False, + include_received_at=True, + ) + + kwargs = api._data_low_level_client.get_channel_data.await_args_list[0].kwargs + assert kwargs["include_received_at"] is True + + @pytest.mark.asyncio + async def test_include_received_at_defaults_to_false(self): + api = _make_api() + api._data_low_level_client = MagicMock() + api._data_low_level_client.get_channel_data = AsyncMock(return_value={}) + + await api.get_data(channels=[_mock_channel(unit="")], show_progress=False) + + kwargs = api._data_low_level_client.get_channel_data.await_args_list[0].kwargs + assert kwargs["include_received_at"] is False diff --git a/python/lib/sift_client/_tests/sift_types/test_channel.py b/python/lib/sift_client/_tests/sift_types/test_channel.py index 226e8bcdc..c6a6f6b88 100644 --- a/python/lib/sift_client/_tests/sift_types/test_channel.py +++ b/python/lib/sift_client/_tests/sift_types/test_channel.py @@ -91,6 +91,7 @@ def test_data_method_calls_get_data(self, mock_channel, mock_client): start_time=datetime(2024, 1, 1, tzinfo=timezone.utc), end_time=datetime(2024, 1, 2, tzinfo=timezone.utc), limit=100, + include_received_at=False, ) assert result == mock_data @@ -112,10 +113,24 @@ def test_data_method_as_arrow(self, mock_channel, mock_client): start_time=None, end_time=None, limit=None, + include_received_at=False, ) mock_client.channels.get_data.assert_not_called() assert result == mock_data + def test_data_method_forwards_include_received_at(self, mock_channel, mock_client): + """data(include_received_at=True) forwards the flag to get_data.""" + mock_channel.data(include_received_at=True) + + assert mock_client.channels.get_data.call_args.kwargs["include_received_at"] is True + + def test_data_method_as_arrow_forwards_include_received_at(self, mock_channel, mock_client): + """data(as_arrow=True, include_received_at=True) forwards the flag to get_data_as_arrow.""" + mock_channel.data(as_arrow=True, include_received_at=True) + + kwargs = mock_client.channels.get_data_as_arrow.call_args.kwargs + assert kwargs["include_received_at"] is True + def test_channel_reference_requires_one_target(self): """ChannelReference must specify exactly one of identifier or calculated_channel.""" with pytest.raises(ValueError, match="exactly one"): @@ -225,6 +240,7 @@ def test_data_method_with_minimal_params(self, mock_channel, mock_client): start_time=None, end_time=None, limit=None, + include_received_at=False, ) assert result == mock_data diff --git a/python/lib/sift_client/resources/channels.py b/python/lib/sift_client/resources/channels.py index 1d8c8d81f..791f429a3 100644 --- a/python/lib/sift_client/resources/channels.py +++ b/python/lib/sift_client/resources/channels.py @@ -268,6 +268,7 @@ async def get_data( page_size: int | None = None, ignore_cache: bool = False, show_progress: bool | None = None, + include_received_at: bool = False, ) -> dict[str, pd.DataFrame]: """Get data for one or more channels. @@ -284,6 +285,10 @@ async def get_data( show_progress: If True, display a progress bar naming each channel as its data is fetched. Defaults to True for sync, False for async. Use ``sift_client.config.show_progress = False`` to disable globally. + include_received_at: If True, each DataFrame gains a + ".sift_received_at" column with the time each + point was received by Sift. Not available for enum or bitfield + channels. Implies ignore_cache (received-at is never cached). Returns: A dictionary mapping channel names to pandas DataFrames containing the channel data. @@ -303,6 +308,7 @@ async def get_data( page_size=page_size, ignore_cache=ignore_cache, show_progress=show_progress, + include_received_at=include_received_at, ) async def get_data_as_arrow( @@ -316,6 +322,7 @@ async def get_data_as_arrow( page_size: int | None = None, ignore_cache: bool = False, show_progress: bool | None = None, + include_received_at: bool = False, ) -> dict[str, pa.Table]: """Get data for one or more channels as pyarrow tables.""" from pyarrow import Table as ArrowTable @@ -330,5 +337,6 @@ async def get_data_as_arrow( page_size=page_size, ignore_cache=ignore_cache, show_progress=show_progress, + include_received_at=include_received_at, ) return {k: ArrowTable.from_pandas(v) for k, v in data.items()} diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index e92fbd35d..709e776b0 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -497,6 +497,7 @@ class ChannelsAPI: page_size: int | None = None, ignore_cache: bool = False, show_progress: bool | None = None, + include_received_at: bool = False, ) -> dict[str, pd.DataFrame]: """Get data for one or more channels. @@ -513,6 +514,10 @@ class ChannelsAPI: show_progress: If True, display a progress bar naming each channel as its data is fetched. Defaults to True for sync, False for async. Use ``sift_client.config.show_progress = False`` to disable globally. + include_received_at: If True, each DataFrame gains a + ".sift_received_at" column with the time each + point was received by Sift. Not available for enum or bitfield + channels. Implies ignore_cache (received-at is never cached). Returns: A dictionary mapping channel names to pandas DataFrames containing the channel data. @@ -530,6 +535,7 @@ class ChannelsAPI: page_size: int | None = None, ignore_cache: bool = False, show_progress: bool | None = None, + include_received_at: bool = False, ) -> dict[str, pa.Table]: """Get data for one or more channels as pyarrow tables.""" ... diff --git a/python/lib/sift_client/sift_types/channel.py b/python/lib/sift_client/sift_types/channel.py index 03e8129bc..60570ad50 100644 --- a/python/lib/sift_client/sift_types/channel.py +++ b/python/lib/sift_client/sift_types/channel.py @@ -312,6 +312,7 @@ def data( end_time: datetime | None = None, limit: int | None = None, as_arrow: bool = False, + include_received_at: bool = False, ): """Retrieve channel data for this channel during the specified run. @@ -321,6 +322,10 @@ def data( end_time: The end time to get data for. limit: The maximum number of data points to return. as_arrow: Whether to return the data as an Arrow table. + include_received_at: If True, include a ".sift_received_at" + column with the time each point was received by Sift. Not + available for enum or bitfield channels. Received-at data is + never cached, so these calls always fetch from the server. Returns: A dict of channel name to pandas DataFrame or Arrow Table object. @@ -332,6 +337,7 @@ def data( start_time=start_time, end_time=end_time, limit=limit, # type: ignore + include_received_at=include_received_at, ) else: data = self.client.channels.get_data( @@ -340,6 +346,7 @@ def data( start_time=start_time, end_time=end_time, limit=limit, # type: ignore + include_received_at=include_received_at, ) return data From 0f337dcc79298d91227df61a78419a13778b3aae Mon Sep 17 00:00:00 2001 From: Wei Lu Date: Thu, 27 Aug 2026 10:49:50 -0700 Subject: [PATCH 2/2] update changelog --- python/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index ee4495977..73a60ada4 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -9,7 +9,7 @@ This project adheres to [Semantic Versioning](http://semver.org/). #### `include_received_at` on `get_data` -`client.channels.get_data(...)`, `get_data_as_arrow(...)`, and `Channel.data(...)` now take `include_received_at`. When set, each returned DataFrame gains a `".sift_received_at"` column with the time each point was received by Sift, alongside the measurement timestamps. Not available for enum or bitfield channels. Received-at data is never cached, so these calls always fetch from the server. +`client.channels.get_data(...)`, `get_data_as_arrow(...)`, and `Channel.data(...)` now take `include_received_at`. When set, each returned DataFrame gains a `".sift_received_at"` column with the time each point was received by Sift, alongside the measurement timestamps. Not available for enum or bitfield channels. ```python data = client.channels.get_data(