fix(records): keep sync PR scoped to sync endpoint

Remove list/filter endpoint surface from the records sync branch and route sync through the shared _list helper for cursor handling.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: andersfylling

cognite/client/_api/data_modeling/records.py

@@ -5,18 +5,15 @@
 from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
 from cognite.client._api_client import APIClient
-from cognite.client.data_classes.data_modeling.instances import InstanceSort
 from cognite.client.data_classes.data_modeling.records import (
-    Record,
     RecordId,
     RecordIdSequence,
-    RecordList,
     RecordSourceSelector,
     RecordTargetUnit,
     RecordTargetUnits,
     RecordWrite,
+    SyncRecord,
     SyncRecordList,
-    TimeRange,
 )
 from cognite.client.data_classes.filters import Filter
 from cognite.client.utils._concurrency import RecordsConcurrencyOperation
@@ -27,9 +24,6 @@
     from cognite.client import AsyncCogniteClient
     from cognite.client.config import ClientConfig
 
-StreamType = Literal["immutable", "mutable"]
-_DEFAULT_STREAM_TYPE: StreamType = "immutable"
-
 
 class RecordsAPI(APIClient):
     def __init__(self, config: ClientConfig, api_version: str | None, cognite_client: AsyncCogniteClient) -> None:
@@ -44,9 +38,7 @@ def __init__(self, config: ClientConfig, api_version: str | None, cognite_client
         "delete": RecordsConcurrencyOperation.WRITE,
     }
 
-    def _get_semaphore(  # type: ignore[override]
-        self, operation: Literal["read", "write", "delete"], stream_type: StreamType
-    ) -> 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(
@@ -69,7 +61,6 @@ async def delete(
         items: RecordId | Sequence[RecordId],
         *,
         stream_id: str,
-        stream_type: StreamType = _DEFAULT_STREAM_TYPE,
         ignore_unknown_ids: Literal[True] = True,
     ) -> None:
         """`Delete records from a stream <https://api-docs.cognite.com/20230101/tag/Records/operation/deleteRecords>`_.
@@ -80,7 +71,6 @@ async def delete(
         Args:
             items (RecordId | Sequence[RecordId]): Records to delete.
             stream_id (str): External ID of the stream to delete from.
-            stream_type (StreamType): Type of the stream ("immutable" or "mutable"). Defaults to "immutable".
             ignore_unknown_ids (Literal[True]): Currently only True is supported
 
         Examples:
@@ -103,15 +93,13 @@ async def delete(
             identifiers=RecordIdSequence.load(items),
             wrap_ids=True,
             resource_path=self._records_url(stream_id),
-            override_semaphore=self._get_semaphore("delete", stream_type),
         )
 
     async def ingest(
         self,
         items: RecordWrite | Sequence[RecordWrite],
         *,
         stream_id: str,
-        stream_type: StreamType = _DEFAULT_STREAM_TYPE,
     ) -> None:
         """`Ingest records into a stream <https://api-docs.cognite.com/20230101/tag/Records/operation/ingestRecords>`_.
 
@@ -123,7 +111,6 @@ async def ingest(
         Args:
             items (RecordWrite | Sequence[RecordWrite]): One or more records to ingest.
             stream_id (str): External ID of the stream to ingest into.
-            stream_type (StreamType): Type of the stream ("immutable" or "mutable"). Defaults to "immutable".
 
         Examples:
 
@@ -158,15 +145,13 @@ async def ingest(
             items=item_list,
             resource_path=self._records_url(stream_id),
             no_response=True,
-            override_semaphore=self._get_semaphore("write", stream_type),
         )
 
     async def upsert(
         self,
         items: RecordWrite | Sequence[RecordWrite],
         *,
         stream_id: str,
-        stream_type: StreamType = _DEFAULT_STREAM_TYPE,
         upsert_mode: Literal["replace"] = "replace",
     ) -> None:
         """`Upsert records into a stream <https://api-docs.cognite.com/20230101/tag/Records/operation/upsertRecords>`_.
@@ -179,7 +164,6 @@ async def upsert(
         Args:
             items (RecordWrite | Sequence[RecordWrite]): One or more records to upsert.
             stream_id (str): External ID of the stream to upsert into.
-            stream_type (StreamType): Type of the stream ("immutable" or "mutable"). Defaults to "immutable".
             upsert_mode (Literal['replace']): How existing records are updated. Currently only ``"replace"`` is supported, which fully replaces the existing record. Defaults to ``"replace"``.
 
         Examples:
@@ -215,97 +199,18 @@ async def upsert(
             items=item_list,
             resource_path=self._records_url(stream_id, "/upsert"),
             no_response=True,
-            override_semaphore=self._get_semaphore("write", stream_type),
-        )
-
-    async def list(
-        self,
-        stream_id: str,
-        *,
-        stream_type: StreamType = _DEFAULT_STREAM_TYPE,
-        last_updated_time: TimeRange | None = None,
-        filter: Filter | None = None,
-        sources: Sequence[RecordSourceSelector] | None = None,
-        target_units: RecordTargetUnits | Sequence[RecordTargetUnit] | 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.
-            stream_type (StreamType): Type of the stream ("immutable" or "mutable"). Defaults to "immutable".
-            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.
-            target_units (RecordTargetUnits | Sequence[RecordTargetUnit] | None): Properties to convert
-                to another unit.
-            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 target_units is not None:
-            other_params["targetUnits"] = self._dump_target_units(target_units)
-        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", stream_type),
         )
 
     async def sync(
         self,
         stream_id: str,
         *,
-        stream_type: StreamType = _DEFAULT_STREAM_TYPE,
         cursor: str | None = None,
         initialize_cursor: str | None = None,
         filter: Filter | None = None,
         sources: Sequence[RecordSourceSelector] | None = None,
         target_units: RecordTargetUnits | Sequence[RecordTargetUnit] | None = None,
-        limit: int | None = None,
+        limit: int = 10,
         include_typing: bool = False,
     ) -> SyncRecordList:
         """`Sync records from a stream <https://api-docs.cognite.com/20230101/tag/Records/operation/syncRecords>`_.
@@ -318,15 +223,14 @@ async def sync(
 
         Args:
             stream_id (str): External ID of the stream to sync.
-            stream_type (StreamType): Type of the stream ("immutable" or "mutable"). Defaults to "immutable".
             cursor (str | None): Resume from a cursor returned by a previous sync call.
             initialize_cursor (str | None): Where to start when no ``cursor`` is given, as a
                 relative duration like ``"7d-ago"``. Ignored when ``cursor`` is set.
             filter (Filter | None): Filter expression (see :mod:`cognite.client.data_classes.filters`).
             sources (Sequence[RecordSourceSelector] | None): Which container properties to return.
             target_units (RecordTargetUnits | Sequence[RecordTargetUnit] | None): Properties to convert
                 to another unit.
-            limit (int | None): Maximum number of records to return in this page (1-1000).
+            limit (int): Maximum number of records to return in this page (1-1000). Defaults to 10.
             include_typing (bool): If True, include property type information on the returned
                 list's ``typing`` attribute.
 
@@ -351,25 +255,26 @@ async def sync(
         self._warning.warn()
         if cursor is not None and initialize_cursor is not None:
             raise ValueError("Provide either 'cursor' or 'initialize_cursor', not both.")
-        body: dict[str, Any] = {}
-        if cursor is not None:
-            body["cursor"] = cursor
-        elif initialize_cursor is not None:
-            body["initializeCursor"] = initialize_cursor
-        if filter is not None:
-            body["filter"] = filter.dump(camel_case_property=False)
+        other_params: dict[str, Any] = {}
+        if initialize_cursor is not None:
+            other_params["initializeCursor"] = initialize_cursor
         if sources is not None:
-            body["sources"] = [source.dump() for source in sources]
+            other_params["sources"] = [source.dump() for source in sources]
         if target_units is not None:
-            body["targetUnits"] = self._dump_target_units(target_units)
-        if limit is not None:
-            body["limit"] = limit
+            other_params["targetUnits"] = self._dump_target_units(target_units)
         if include_typing:
-            body["includeTyping"] = True
+            other_params["includeTyping"] = True
 
-        response = await self._post(
+        return await self._list(
+            list_cls=SyncRecordList,
+            resource_cls=SyncRecord,
+            method="POST",
+            resource_path=self._records_url(stream_id),
             url_path=self._records_url(stream_id, "/sync"),
-            json=body,
-            semaphore=self._get_semaphore("read", stream_type),
+            limit=limit,
+            filter=filter.dump(camel_case_property=False) if isinstance(filter, Filter) else filter,
+            other_params=other_params,
+            initial_cursor=cursor,
+            settings_forcing_raw_response_loading=["records_sync_cursor"],
+            override_semaphore=self._get_semaphore("read"),
         )
-        return SyncRecordList._load_response(response.json())