feat(records): add list (filter) endpoint with read data classes

Add RecordsAPI.list, a cursorless POST to /streams/{streamId}/records/filter
returning a RecordList (max 1000 records). Introduces the records read model:

- Record / RecordList read data classes (clean CogniteResource, no RecordId
  multiple-inheritance; RecordList carries optional `typing`).
- TimeRange (gte/gt/lte/lt) for last_updated_time and RecordSourceSelector for
  source/property selection; reuse the data-modeling Filter DSL, InstanceSort,
  and TypeInformation.
- Add a READ op to RecordsConcurrencyOperation + a read semaphore to the
  records concurrency config (reads no longer borrow the write semaphore).
- Make _records_url append the path suffix literally (encode only stream_id)
  so "/filter" isn't percent-encoded.

Also register the records API module in the docstring-example doctest runner.
target_units is intentionally deferred to a follow-up.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: andersfylling

cognite/client/_api/data_modeling/records.py

@@ -2,10 +2,19 @@
 
 import asyncio
 from collections.abc import Sequence
-from typing import TYPE_CHECKING, ClassVar, Literal
+from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
 from cognite.client._api_client import APIClient
-from cognite.client.data_classes.data_modeling.records import RecordId, RecordIdSequence, RecordWrite
+from cognite.client.data_classes.data_modeling.instances import InstanceSort
+from cognite.client.data_classes.data_modeling.records import (
+    RecordId,
+    RecordIdSequence,
+    RecordList,
+    RecordSourceSelector,
+    RecordWrite,
+    TimeRange,
+)
+from cognite.client.data_classes.filters import Filter
 from cognite.client.utils._concurrency import RecordsConcurrencyOperation
 from cognite.client.utils._experimental import FeaturePreviewWarning
 from cognite.client.utils._url import interpolate_and_url_encode
@@ -23,19 +32,22 @@ def __init__(self, config: ClientConfig, api_version: str | None, cognite_client
         )
 
     _OPERATION_TO_RATE_LIMIT: ClassVar[dict[str, RecordsConcurrencyOperation]] = {
+        "read": RecordsConcurrencyOperation.READ,
         "write": RecordsConcurrencyOperation.WRITE,
         "delete": RecordsConcurrencyOperation.WRITE,
     }
 
-    def _get_semaphore(self, operation: Literal["write", "delete"]) -> asyncio.BoundedSemaphore:
+    def _get_semaphore(self, operation: Literal["read", "write", "delete"]) -> asyncio.BoundedSemaphore:
         from cognite.client import global_config
 
         return global_config.concurrency_settings.records._semaphore_factory(
             self._OPERATION_TO_RATE_LIMIT[operation], project=self._cognite_client.config.project
         )
 
     def _records_url(self, stream_id: str, suffix: str = "") -> str:
-        return interpolate_and_url_encode("/streams/{}/records{}", stream_id, suffix)
+        # Encode only stream_id; the suffix is a literal path segment (e.g. "/filter"),
+        # so it must not be percent-encoded.
+        return interpolate_and_url_encode("/streams/{}/records", stream_id) + suffix
 
     async def delete(
         self,
@@ -127,3 +139,69 @@ async def ingest(
             resource_path=self._records_url(stream_id),
             no_response=True,
         )
+
+    async def list(
+        self,
+        stream_id: str,
+        *,
+        last_updated_time: TimeRange | None = None,
+        filter: Filter | None = None,
+        sources: Sequence[RecordSourceSelector] | None = None,
+        sort: Sequence[InstanceSort] | InstanceSort | None = None,
+        limit: int = 10,
+        include_typing: bool = False,
+    ) -> RecordList:
+        """`Filter records in a stream <https://api-docs.cognite.com/20230101/tag/Records/operation/filterRecords>`_.
+
+        Returns records matching the given filters, sorted by ``lastUpdatedTime`` unless a custom
+        ``sort`` is given. This endpoint is not cursor-paged: it returns at most ``limit`` records
+        (max 1000). To page over a large time window, issue multiple calls with partitioned
+        ``last_updated_time`` ranges.
+
+        Args:
+            stream_id (str): External ID of the stream to query.
+            last_updated_time (TimeRange | None): Filter by last-updated time. **Required for
+                immutable streams** (must include a lower bound).
+            filter (Filter | None): Filter expression (see :mod:`cognite.client.data_classes.filters`).
+            sources (Sequence[RecordSourceSelector] | None): Which container properties to return.
+            sort (Sequence[InstanceSort] | InstanceSort | None): Sort specification(s); up to 5.
+            limit (int): Maximum number of records to return (1-1000). Defaults to 10.
+            include_typing (bool): If True, include property type information on the returned
+                list's ``typing`` attribute.
+
+        Returns:
+            RecordList: The matching records.
+
+        Examples:
+
+            List records updated since a given timestamp:
+
+                >>> from cognite.client import CogniteClient
+                >>> from cognite.client.data_classes.data_modeling.records import TimeRange
+                >>> client = CogniteClient()
+                >>> res = client.data_modeling.records.list(
+                ...     stream_id="my-stream",
+                ...     last_updated_time=TimeRange(gt=1705341600000),
+                ...     limit=100,
+                ... )
+        """
+        self._warning.warn()
+        body: dict[str, Any] = {"limit": limit}
+        if last_updated_time is not None:
+            body["lastUpdatedTime"] = last_updated_time.dump()
+        if filter is not None:
+            body["filter"] = filter.dump()
+        if sources is not None:
+            body["sources"] = [source.dump() for source in sources]
+        if sort is not None:
+            sort_list = [sort] if isinstance(sort, InstanceSort) else list(sort)
+            body["sort"] = [spec.dump() for spec in sort_list]
+        if include_typing:
+            body["includeTyping"] = True
+
+        response = await self._post(
+            url_path=self._records_url(stream_id, "/filter"),
+            json=body,
+            semaphore=self._get_semaphore("read"),
+        )
+        return RecordList._load_raw_api_response([response.json()])

cognite/client/_sync_api/data_modeling/records.py

@@ -120,11 +120,15 @@
     UnionAll,
 )
 from cognite.client.data_classes.data_modeling.records import (
+    Record,
     RecordContainerId,
     RecordId,
+    RecordList,
     RecordSource,
+    RecordSourceSelector,
     RecordWrite,
     RecordWriteList,
+    TimeRange,
 )
 from cognite.client.data_classes.data_modeling.spaces import Space, SpaceApply, SpaceApplyList, SpaceList
 from cognite.client.data_classes.data_modeling.streams import (
@@ -249,9 +253,12 @@
     "Query",
     "QueryResult",
     "QuerySync",
+    "Record",
     "RecordContainerId",
     "RecordId",
+    "RecordList",
     "RecordSource",
+    "RecordSourceSelector",
     "RecordWrite",
     "RecordWriteList",
     "RequiresConstraint",
@@ -279,6 +286,7 @@
     "StreamWrite",
     "SubscriptionContext",
     "Text",
+    "TimeRange",
     "TimeSeriesReference",
     "Timestamp",
     "TranslatedQuery",