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

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: andersfylling

cognite/client/_api/data_modeling/records.py

@@ -2,10 +2,20 @@
 
 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 (
+    Record,
+    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
@@ -14,7 +24,6 @@
     from cognite.client import AsyncCogniteClient
     from cognite.client.config import ClientConfig
 
-
 class RecordsAPI(APIClient):
     def __init__(self, config: ClientConfig, api_version: str | None, cognite_client: AsyncCogniteClient) -> None:
         super().__init__(config, api_version, cognite_client)
@@ -23,11 +32,14 @@ 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(  # type: ignore[override]
+        self, operation: Literal["read", "write", "delete"]
+    ) -> asyncio.BoundedSemaphore:
         from cognite.client import global_config
 
         return global_config.concurrency_settings.records._semaphore_factory(
@@ -76,6 +88,7 @@ async def delete(
             identifiers=RecordIdSequence.load(items),
             wrap_ids=True,
             resource_path=self._records_url(stream_id),
+            override_semaphore=self._get_semaphore("delete"),
         )
 
     async def ingest(
@@ -128,6 +141,7 @@ async def ingest(
             items=item_list,
             resource_path=self._records_url(stream_id),
             no_response=True,
+            override_semaphore=self._get_semaphore("write"),
         )
 
     async def upsert(
@@ -182,4 +196,73 @@ async def upsert(
             items=item_list,
             resource_path=self._records_url(stream_id, "/upsert"),
             no_response=True,
+            override_semaphore=self._get_semaphore("write"),
+        )
+
+    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.
+
+        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()
+        other_params: dict[str, Any] = {}
+        if last_updated_time is not None:
+            other_params["lastUpdatedTime"] = last_updated_time.dump()
+        if sources is not None:
+            other_params["sources"] = [source.dump() for source in sources]
+        if sort is not None:
+            sort_list = [sort] if isinstance(sort, InstanceSort) else list(sort)
+            other_params["sort"] = [spec.dump() for spec in sort_list]
+        if include_typing:
+            other_params["includeTyping"] = True
+
+        return await self._list(
+            list_cls=RecordList,
+            resource_cls=Record,
+            method="POST",
+            resource_path=self._records_url(stream_id),
+            url_path=self._records_url(stream_id, "/filter"),
+            limit=limit,
+            filter=filter.dump(camel_case_property=False) if isinstance(filter, Filter) else filter,
+            other_params=other_params,
+            settings_forcing_raw_response_loading=[f"{include_typing=}"] if include_typing else None,
+            override_semaphore=self._get_semaphore("read"),
         )

cognite/client/_api/hosted_extractors/destinations.py

@@ -65,7 +65,7 @@ async def __call__(
             yield item
 
     @overload
-    async def retrieve(self, external_ids: str, ignore_unknown_ids: bool = False) -> Destination | None: ...
+    async def retrieve(self, external_ids: str, ignore_unknown_ids: bool = False) -> Destination: ...
 
     @overload
     async def retrieve(
@@ -74,15 +74,16 @@ async def retrieve(
 
     async def retrieve(
         self, external_ids: str | SequenceNotStr[str], ignore_unknown_ids: bool = False
-    ) -> Destination | DestinationList | None:
+    ) -> Destination | DestinationList:
         """`Retrieve one or more destinations <https://api-docs.cognite.com/20230101/tag/Destinations/operation/retrieve_destinations>`_.
 
         Args:
             external_ids (str | SequenceNotStr[str]): The external ID provided by the client. Must be unique for the resource type.
             ignore_unknown_ids (bool): Ignore external IDs that are not found
 
+
         Returns:
-            Destination | DestinationList | None: Requested destinations
+            Destination | DestinationList: Requested destinations
 
         Examples:
 

cognite/client/_api/hosted_extractors/jobs.py

@@ -75,15 +75,15 @@ async def retrieve(self, external_ids: SequenceNotStr[str], ignore_unknown_ids:
 
     async def retrieve(
         self, external_ids: str | SequenceNotStr[str], ignore_unknown_ids: bool = False
-    ) -> Job | JobList | None:
+    ) -> Job | None | JobList:
         """`Retrieve one or more jobs <https://api-docs.cognite.com/20230101/tag/Jobs/operation/retrieve_jobs>`_.
 
         Args:
             external_ids (str | SequenceNotStr[str]): The external ID provided by the client. Must be unique for the job type.
             ignore_unknown_ids (bool): Ignore external IDs that are not found
 
         Returns:
-            Job | JobList | None: Requested jobs
+            Job | None | JobList: Requested jobs
 
         Examples:
 

cognite/client/_api/hosted_extractors/mappings.py

@@ -60,22 +60,23 @@ async def __call__(
             yield item
 
     @overload
-    async def retrieve(self, external_ids: str, ignore_unknown_ids: bool = False) -> Mapping | None: ...
+    async def retrieve(self, external_ids: str, ignore_unknown_ids: bool = False) -> Mapping: ...
 
     @overload
     async def retrieve(self, external_ids: SequenceNotStr[str], ignore_unknown_ids: bool = False) -> MappingList: ...
 
     async def retrieve(
         self, external_ids: str | SequenceNotStr[str], ignore_unknown_ids: bool = False
-    ) -> Mapping | MappingList | None:
+    ) -> Mapping | MappingList:
         """`Retrieve one or more mappings <https://api-docs.cognite.com/20230101/tag/Mappings/operation/retrieve_mappings>`_.
 
         Args:
             external_ids (str | SequenceNotStr[str]): The external ID provided by the client. Must be unique for the resource type.
             ignore_unknown_ids (bool): Ignore external IDs that are not found
 
+
         Returns:
-            Mapping | MappingList | None: Requested mappings
+            Mapping | MappingList: Requested mappings
 
         Examples:
 

cognite/client/_api/hosted_extractors/sources.py

@@ -61,22 +61,22 @@ async def __call__(
             yield item
 
     @overload
-    async def retrieve(self, external_ids: str, ignore_unknown_ids: bool = False) -> Source | None: ...
+    async def retrieve(self, external_ids: str, ignore_unknown_ids: bool = False) -> Source: ...
 
     @overload
     async def retrieve(self, external_ids: SequenceNotStr[str], ignore_unknown_ids: bool = False) -> SourceList: ...
 
     async def retrieve(
         self, external_ids: str | SequenceNotStr[str], ignore_unknown_ids: bool = False
-    ) -> Source | SourceList | None:
+    ) -> Source | SourceList:
         """`Retrieve one or more sources <https://api-docs.cognite.com/20230101/tag/Sources/operation/retrieve_sources>`_.
 
         Args:
             external_ids (str | SequenceNotStr[str]): The external ID provided by the client. Must be unique for the resource type.
             ignore_unknown_ids (bool): Ignore external IDs that are not found rather than throw an exception.
 
         Returns:
-            Source | SourceList | None: Requested sources
+            Source | SourceList: Requested sources
 
         Examples:
 

cognite/client/_api/limits.py

@@ -50,7 +50,7 @@ async def retrieve(self, id: str) -> Limit | None:
         """
         self._warning.warn()
 
-        headers = self._alpha_version_header()
+        headers = {"cdf-version": f"{self._config.api_subversion}-alpha"}
 
         return await self._retrieve(
             identifier=LimitId(id),
@@ -87,7 +87,7 @@ async def list(self, filter: Prefix | None = None, limit: int | None = DEFAULT_L
         """
         self._warning.warn()
 
-        headers = self._alpha_version_header()
+        headers = {"cdf-version": f"{self._config.api_subversion}-alpha"}
 
         return await self._list(
             method="GET" if filter is None else "POST",