Python(fix): cache boundary resolution to include ns

python/CHANGELOG.md

@@ -37,6 +37,9 @@ CsvDataColumn(
 
 Setting `enum_types` on a non-enum data type raises a validation error. Multi-channel Parquet single-channel-per-row imports don't support enums, since channel configs there are derived per row.
 
+### Bugfixes
+- Fix `channels.get_data` printing a `UserWarning: Discarding nonzero nanoseconds in conversion` once per channel on each run-scoped fetch. The channel data cache stored segment bounds as `datetime`, which truncated the nanosecond timestamps the data already carries; they are now `pd.Timestamp`. `start_time` and `end_time` on `get_data` and `get_data_as_arrow` also accept a `pd.Timestamp` now, and its nanoseconds survive to the wire, where the service filters on them.
+
 ## [v0.20.0] - August 25, 2026
 
 ### What's New

python/lib/sift_client/_internal/low_level_wrappers/data.py

@@ -38,7 +38,7 @@
 REQUEST_BATCH_SIZE = 1
 
 
-TimeRange = Tuple[datetime, datetime]
+TimeRange = Tuple[pd.Timestamp, pd.Timestamp]
 
 
 class SegmentRef(BaseModel):
@@ -48,12 +48,17 @@ class SegmentRef(BaseModel):
     queried this range and the wire returned no data". Empty refs
     contribute to coverage (so a repeat of a known-empty range
     doesn't hit the wire).
+
+    Bounds are ``pd.Timestamp``: a ``datetime`` truncates to
+    microseconds, which would under-claim the first sample by up to
+    999 ns. The adapter's methods normalize what they're handed, so
+    callers can still pass a plain ``datetime``.
     """
 
-    model_config = ConfigDict(frozen=True)
+    model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
     seg_id: int | None
-    start_time: datetime
-    end_time: datetime
+    start_time: pd.Timestamp
+    end_time: pd.Timestamp
 
 
 class SegmentIndex(BaseModel):
@@ -159,8 +164,8 @@ def get_range(
         self,
         channel_id: str,
         run_id: str | None,
-        start_time: datetime,
-        end_time: datetime,
+        start_time: pd.Timestamp | datetime,
+        end_time: pd.Timestamp | datetime,
     ) -> tuple[pd.DataFrame | None, list[TimeRange]]:
         """Return cached data covering ``[start_time, end_time]`` plus gaps.
 
@@ -176,6 +181,8 @@ def get_range(
             query window not covered by any present segment.
             ``gaps == []`` means the cache fully covers the request.
         """
+        start_time = to_timestamp_nanos(start_time)
+        end_time = to_timestamp_nanos(end_time)
         idx = self._load_index(channel_id, run_id)
         if idx is None or not idx.segments:
             return None, [(start_time, end_time)]
@@ -226,8 +233,8 @@ def put_segment(
         channel_id: str,
         run_id: str | None,
         data: pd.DataFrame,
-        start_time: datetime,
-        end_time: datetime,
+        start_time: pd.Timestamp | datetime,
+        end_time: pd.Timestamp | datetime,
     ) -> None:
         """Write a new segment and update the index.
 
@@ -254,6 +261,8 @@ def put_segment(
         puts skip the segment write entirely and update the index
         directly.
         """
+        start_time = to_timestamp_nanos(start_time)
+        end_time = to_timestamp_nanos(end_time)
         idx = self._load_index(channel_id, run_id) or SegmentIndex()
 
         if len(data) == 0:
@@ -486,8 +495,8 @@ def _replace_index_and_drop_segments(
 
     @staticmethod
     def _compute_gaps(
-        query_start: datetime,
-        query_end: datetime,
+        query_start: pd.Timestamp,
+        query_end: pd.Timestamp,
         covered: list[TimeRange],
     ) -> list[TimeRange]:
         """Sub-ranges of ``[query_start, query_end]`` not in ``covered``.
@@ -566,8 +575,8 @@ async def _get_data_impl(
         *,
         channel_ids: list[str],
         run_id: str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime,
         page_size: int | None = None,
         page_token: str | None = None,
         order_by: str | None = None,
@@ -642,8 +651,8 @@ def _update_cache(
         *,
         channel_data: dict[str, pd.DataFrame],
         fetched_ranges_per_channel: dict[str, list[TimeRange]],
-        start_time: datetime,
-        end_time: datetime,
+        start_time: pd.Timestamp,
+        end_time: pd.Timestamp,
         run_id: str | None = None,
     ):
         """Write each channel's fresh data or empty-ref as a new segment.
@@ -720,12 +729,13 @@ def _update_cache(
             # started yet). For unscoped queries, claim the full
             # requested range so a follow-up of the same range hits.
             seg_end = end_time
+            seg_start: pd.Timestamp
             if run_id:
                 # ``combined.index`` is a ``DatetimeIndex`` (built from
                 # the wire's nanosecond timestamps), so ``index[0]`` is
                 # always ``pd.Timestamp`` at runtime; pandas-stubs types
                 # it as the wider ``Scalar`` union.
-                seg_start = cast("pd.Timestamp", combined.index[0]).to_pydatetime()
+                seg_start = cast("pd.Timestamp", combined.index[0])
             else:
                 seg_start = start_time
 
@@ -767,8 +777,8 @@ async def get_channel_data(
         *,
         channels: list[Channel],
         run_id: str | None = None,
-        start_time: datetime | None = None,
-        end_time: datetime | None = None,
+        start_time: pd.Timestamp | datetime | None = None,
+        end_time: pd.Timestamp | datetime | None = None,
         max_results: int | None = None,
         page_size: int | None = None,
         ignore_cache: bool = False,
@@ -777,8 +787,8 @@ async def get_channel_data(
         """Get the data for a channel during a run."""
         ret_data: dict[str, pd.DataFrame] = {}
         # No data will be returned if end_time is not provided.
-        start_time = start_time or datetime.fromtimestamp(0, tz=timezone.utc)
-        end_time = end_time or datetime.now(timezone.utc)
+        query_start = to_timestamp_nanos(start_time or datetime.fromtimestamp(0, tz=timezone.utc))
+        query_end = to_timestamp_nanos(end_time or datetime.now(timezone.utc))
 
         self._update_name_id_map(channels)
 
@@ -803,9 +813,11 @@ async def get_channel_data(
             assert cid is not None
             if ignore_cache:
                 cached_data: pd.DataFrame | None = None
-                gaps: list[TimeRange] = [(start_time, end_time)]
+                gaps: list[TimeRange] = [(query_start, query_end)]
             else:
-                cached_data, gaps = self.channel_cache.get_range(cid, run_id, start_time, end_time)
+                cached_data, gaps = self.channel_cache.get_range(
+                    cid, run_id, query_start, query_end
+                )
 
             if cached_data is not None:
                 cached_points += int(cached_data.size)
@@ -820,7 +832,7 @@ async def get_channel_data(
             if not gaps:
                 continue
             fetched_ranges_per_channel.setdefault(cid, []).extend(gaps)
-            if len(gaps) == 1 and gaps[0] == (start_time, end_time):
+            if len(gaps) == 1 and gaps[0] == (query_start, query_end):
                 fully_uncached.append(cid)
             else:
                 partial_gaps.append((cid, gaps))
@@ -841,8 +853,8 @@ async def get_channel_data(
                     {
                         "channel_ids": batch,
                         "run_id": run_id,
-                        "start_time": start_time,
-                        "end_time": end_time,
+                        "start_time": query_start,
+                        "end_time": query_end,
                     },
                     ", ".join(id_to_name.get(cid, cid) for cid in batch),
                 )
@@ -963,8 +975,8 @@ async def _tick(kwargs: dict[str, Any], label: str) -> Any:
             self._update_cache(
                 channel_data=ret_data,
                 fetched_ranges_per_channel=fetched_ranges_per_channel,
-                start_time=start_time,
-                end_time=end_time,
+                start_time=query_start,
+                end_time=query_end,
                 run_id=run_id,
             )
 

python/lib/sift_client/_internal/low_level_wrappers/exports.py

@@ -19,7 +19,7 @@
 from sift.exports.v1.exports_pb2_grpc import ExportServiceStub
 
 from sift_client._internal.low_level_wrappers.base import LowLevelClientBase
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client.sift_types.calculated_channel import CalculatedChannel, CalculatedChannelCreate
 from sift_client.transport import WithGrpcClient
 
@@ -123,25 +123,25 @@ async def export_data(
         if run_ids is not None:
             runs_and_time_range = RunsAndTimeRange(run_ids=run_ids)
             if start_time:
-                runs_and_time_range.start_time.CopyFrom(to_pb_timestamp(start_time))
+                runs_and_time_range.start_time.CopyFrom(to_timestamp_pb(start_time))
             if stop_time:
-                runs_and_time_range.stop_time.CopyFrom(to_pb_timestamp(stop_time))
+                runs_and_time_range.stop_time.CopyFrom(to_timestamp_pb(stop_time))
             request.runs_and_time_range.CopyFrom(runs_and_time_range)
 
         if asset_ids is not None:
             assets_and_time_range = AssetsAndTimeRange(asset_ids=asset_ids)
             if start_time:
-                assets_and_time_range.start_time.CopyFrom(to_pb_timestamp(start_time))
+                assets_and_time_range.start_time.CopyFrom(to_timestamp_pb(start_time))
             if stop_time:
-                assets_and_time_range.stop_time.CopyFrom(to_pb_timestamp(stop_time))
+                assets_and_time_range.stop_time.CopyFrom(to_timestamp_pb(stop_time))
             request.assets_and_time_range.CopyFrom(assets_and_time_range)
 
         if run_ids is None and asset_ids is None:
             time_range = TimeRange()
             if start_time:
-                time_range.start_time.CopyFrom(to_pb_timestamp(start_time))
+                time_range.start_time.CopyFrom(to_timestamp_pb(start_time))
             if stop_time:
-                time_range.stop_time.CopyFrom(to_pb_timestamp(stop_time))
+                time_range.stop_time.CopyFrom(to_timestamp_pb(stop_time))
             request.time_range.CopyFrom(time_range)
 
         response = await self._grpc_client.get_stub(ExportServiceStub).ExportData(request)

python/lib/sift_client/_internal/low_level_wrappers/rules.py

@@ -54,7 +54,7 @@
 
 from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
 from sift_client._internal.low_level_wrappers.jobs import JobsLowLevelClient
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client._internal.util.util import count_non_none
 from sift_client.sift_types.rule import (
     Rule,
@@ -635,13 +635,13 @@ async def evaluate_rules(
             if run_id:
                 kwargs["run_time_range"] = RunTimeRange(
                     run=run_id,  # type: ignore
-                    start_time=to_pb_timestamp(start_time),
-                    end_time=to_pb_timestamp(end_time),  # type: ignore
+                    start_time=to_timestamp_pb(start_time),
+                    end_time=to_timestamp_pb(end_time),  # type: ignore
                 )
                 kwargs["assets"] = AssetsTimeRange(
                     assets={"ids": {"ids": asset_ids}},  # type: ignore
-                    start_time=to_pb_timestamp(start_time),
-                    end_time=to_pb_timestamp(end_time),
+                    start_time=to_timestamp_pb(start_time),
+                    end_time=to_timestamp_pb(end_time),
                 )
         elif run_id:
             kwargs["run"] = ResourceIdentifier(id=run_id)

python/lib/sift_client/_internal/low_level_wrappers/runs.py

@@ -19,7 +19,7 @@
 from sift.runs.v2.runs_pb2_grpc import RunServiceStub
 
 from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase
-from sift_client._internal.util.timestamp import to_pb_timestamp
+from sift_client._internal.time import to_timestamp_pb
 from sift_client.sift_types.run import Run, RunCreate, RunUpdate
 from sift_client.transport import WithGrpcClient
 
@@ -212,8 +212,8 @@ async def create_adhoc_run(
         request = CreateAdhocRunRequest(
             name=name,
             description=description or "",
-            start_time=to_pb_timestamp(start_time) if start_time else None,
-            stop_time=to_pb_timestamp(stop_time) if stop_time else None,
+            start_time=to_timestamp_pb(start_time) if start_time else None,
+            stop_time=to_timestamp_pb(stop_time) if stop_time else None,
             asset_ids=asset_ids,
             tags=tag_names,
             metadata=metadata_dict_to_proto(metadata) if metadata else None,

python/lib/sift_client/_internal/time.py

@@ -31,15 +31,26 @@ def to_timestamp_nanos(arg: TimestampPb | pd.Timestamp | datetime | str | int) -
         return cast("pd.Timestamp", pd.Timestamp(arg))
 
 
-def to_timestamp_pb(arg: datetime | str | int | float) -> TimestampPb:
+def to_timestamp_pb(arg: pd.Timestamp | datetime | str | int | float) -> TimestampPb:
     """
-    Mainly used for testing at the moment. If using this for non-testing purposes
-    should probably make this more robust and support nano-second precision.
+    Converts a variety of time-types to a protobuf timestamp.
+
+    A ``pd.Timestamp`` keeps its nanoseconds. Every other input has
+    microsecond resolution at best, since ``TimestampPb.FromDatetime``
+    reads ``microsecond`` and nothing finer. A naive input is read as
+    UTC on either path.
     """
 
     ts = TimestampPb()
 
-    if isinstance(arg, datetime):
+    if isinstance(arg, pd.Timestamp) and arg.nanosecond:
+        # A nonzero sub-microsecond remainder is exactly what
+        # ``FromDatetime`` drops, and it implies nanosecond units, so
+        # ``value`` can't overflow here. Coarser units reach dates
+        # outside the ns range where it would.
+        ts.FromNanoseconds(arg.value)
+        return ts
+    elif isinstance(arg, datetime):
         ts.FromDatetime(arg)
         return ts
     elif isinstance(arg, (int, float)):