feat(records): sync resume method

Separate initial sync from cursor-based continuation while sharing the request assembly and pagination through the internal _sync helper.

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

Author: andersfylling

cognite/client/_api/data_modeling/records.py

@@ -51,6 +51,42 @@ def _dump_target_units(target_units: RecordTargetUnits | Sequence[RecordTargetUn
             return target_units.dump()
         return RecordTargetUnits(properties=target_units).dump()
 
+    async def _sync(
+        self,
+        stream_id: str,
+        *,
+        filter: Filter | None = None,
+        sources: Sequence[RecordSourceSelector] | None = None,
+        target_units: RecordTargetUnits | Sequence[RecordTargetUnit] | None = None,
+        limit: int = 10,
+        include_typing: bool = False,
+        initialize_cursor: str | None = None,
+        cursor: str | None = None,
+    ) -> SyncRecordList:
+        other_params: dict[str, Any] = {}
+        if initialize_cursor is not None:
+            other_params["initializeCursor"] = initialize_cursor
+        if sources is not None:
+            other_params["sources"] = [source.dump() for source in sources]
+        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=SyncRecordList,
+            resource_cls=SyncRecord,
+            method="POST",
+            resource_path=self._records_url(stream_id),
+            url_path=self._records_url(stream_id, "/sync"),
+            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"),
+        )
+
     async def delete(
         self,
         items: RecordId | Sequence[RecordId],
@@ -200,8 +236,7 @@ async def sync(
         self,
         stream_id: str,
         *,
-        cursor: str | None = None,
-        initialize_cursor: str | None = None,
+        initialize_cursor: str,
         filter: Filter | None = None,
         sources: Sequence[RecordSourceSelector] | None = None,
         target_units: RecordTargetUnits | Sequence[RecordTargetUnit] | None = None,
@@ -210,17 +245,14 @@ async def sync(
     ) -> SyncRecordList:
         """`Sync records from a stream <https://api-docs.cognite.com/20230101/tag/Records/operation/syncRecords>`_.
 
-        Returns the next page of the change feed (new, updated and deleted records). Provide exactly
-        one of ``cursor`` (to resume a previous position) or ``initialize_cursor`` (to start from a
-        relative time such as ``"7d-ago"``). Persist the returned :attr:`SyncRecordList.cursor` and
-        pass it as ``cursor`` on the next call to continue; :attr:`SyncRecordList.has_next` indicates
-        whether more changes are immediately available.
+        Returns the first page of the change feed (new, updated and deleted records). Provide
+        ``initialize_cursor`` to start from a relative time such as ``"7d-ago"``. Persist the returned
+        :attr:`SyncRecordList.cursor` and pass it to :meth:`sync_resume` on the next call to continue;
+        :attr:`SyncRecordList.has_next` indicates whether more changes are immediately available.
 
         Args:
             stream_id (str): External ID of the stream to sync.
-            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.
+            initialize_cursor (str): Where to start, as a relative duration like ``"7d-ago"``.
             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
@@ -243,33 +275,55 @@ async def sync(
                 ... )
                 >>> for record in page:
                 ...     pass  # process record; record.status is created/updated/deleted
-                >>> next_page = client.data_modeling.records.sync(
+                >>> next_page = client.data_modeling.records.sync_resume(
                 ...     stream_id="my-stream", cursor=page.cursor
                 ... )
         """
         self._warning.warn()
-        if cursor is not None and initialize_cursor is not None:
-            raise ValueError("Provide either 'cursor' or 'initialize_cursor', not both.")
-        other_params: dict[str, Any] = {}
-        if initialize_cursor is not None:
-            other_params["initializeCursor"] = initialize_cursor
-        if sources is not None:
-            other_params["sources"] = [source.dump() for source in sources]
-        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._sync(
+            stream_id=stream_id,
+            initialize_cursor=initialize_cursor,
+            limit=limit,
+            filter=filter,
+            sources=sources,
+            target_units=target_units,
+            include_typing=include_typing,
+        )
 
-        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"),
+    async def sync_resume(
+        self,
+        stream_id: str,
+        *,
+        cursor: str,
+        filter: Filter | None = None,
+        sources: Sequence[RecordSourceSelector] | None = None,
+        target_units: RecordTargetUnits | Sequence[RecordTargetUnit] | None = None,
+        limit: int = 10,
+        include_typing: bool = False,
+    ) -> SyncRecordList:
+        """Resume syncing records from a stream using a cursor from :meth:`sync` or :meth:`sync_resume`.
+
+        Args:
+            stream_id (str): External ID of the stream to sync.
+            cursor (str): Resume from a cursor returned by a previous sync call.
+            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): 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.
+
+        Returns:
+            SyncRecordList: One page of change records, with ``cursor`` and ``has_next`` set.
+        """
+        self._warning.warn()
+        return await self._sync(
+            stream_id=stream_id,
+            cursor=cursor,
             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"),
+            filter=filter,
+            sources=sources,
+            target_units=target_units,
+            include_typing=include_typing,
         )

cognite/client/_sync_api/data_modeling/records.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from typing import Any, Literal
 
@@ -54,12 +54,12 @@ class RecordSource(CogniteResource):
 
     Args:
         source (RecordContainerId): Reference to the container.
-        properties (dict[str, Any]): The data to write to the source container.
+        properties (Mapping[str, Any]): The data to write to the source container.
     """
 
-    def __init__(self, source: RecordContainerId, properties: dict[str, Any]) -> None:
+    def __init__(self, source: RecordContainerId, properties: Mapping[str, Any]) -> None:
         self.source = source
-        self.properties = properties
+        self.properties = dict(properties)
 
     @classmethod
     def _load(cls, resource: dict[str, Any]) -> Self:
@@ -83,13 +83,13 @@ class RecordWrite(WriteableCogniteResource["RecordWrite"]):
     Args:
         space (str): Space the record belongs to.
         external_id (str): External ID of the record (1-256 chars, no null bytes).
-        sources (list[RecordSource]): Container property values to write (1-100 sources).
+        sources (Sequence[RecordSource]): Container property values to write (1-100 sources).
     """
 
-    def __init__(self, space: str, external_id: str, sources: list[RecordSource]) -> None:
+    def __init__(self, space: str, external_id: str, sources: Sequence[RecordSource]) -> None:
         self.space = space
         self.external_id = external_id
-        self.sources = sources
+        self.sources = list(sources)
 
     def as_id(self) -> RecordId:
         return RecordId(space=self.space, external_id=self.external_id)
@@ -127,12 +127,12 @@ class RecordSourceSelector(CogniteResource):
 
     Args:
         source (RecordContainerId): The container to select properties from.
-        properties (list[str]): Property identifiers to return; use ``["*"]`` to return all.
+        properties (Sequence[str]): Property identifiers to return; use ``["*"]`` to return all.
     """
 
-    def __init__(self, source: RecordContainerId, properties: list[str]) -> None:
+    def __init__(self, source: RecordContainerId, properties: Sequence[str]) -> None:
         self.source = source
-        self.properties = properties
+        self.properties = list(properties)
 
     @classmethod
     def _load(cls, resource: dict[str, Any]) -> Self:
@@ -211,7 +211,7 @@ class SyncRecord(WriteableCogniteResource["RecordWrite"]):
         created_time (int): Creation time in milliseconds since epoch.
         last_updated_time (int): Last updated time in milliseconds since epoch.
         status (Literal['created', 'updated', 'deleted']): The record's change status.
-        properties (dict[str, dict[str, dict[str, Any]]] | None): Property values (absent for
+        properties (Mapping[str, Mapping[str, Mapping[str, Any]]] | None): Property values (absent for
             deleted tombstones).
     """
 
@@ -222,14 +222,21 @@ def __init__(
         created_time: int,
         last_updated_time: int,
         status: Literal["created", "updated", "deleted"],
-        properties: dict[str, dict[str, dict[str, Any]]] | None = None,
+        properties: Mapping[str, Mapping[str, Mapping[str, Any]]] | None = None,
     ) -> None:
         self.space = space
         self.external_id = external_id
         self.created_time = created_time
         self.last_updated_time = last_updated_time
         self.status = status
-        self.properties = properties
+        self.properties = (
+            {
+                space_key: {container: dict(values) for container, values in containers.items()}
+                for space_key, containers in properties.items()
+            }
+            if properties is not None
+            else None
+        )
 
     @classmethod
     def _load(cls, resource: dict[str, Any]) -> Self:
@@ -275,7 +282,7 @@ class SyncRecordList(CogniteResourceList[SyncRecord]):
 
     Args:
         resources (Sequence[SyncRecord]): The records in this page.
-        cursor (str | None): Cursor to pass as ``cursor`` to the next ``sync`` call to resume
+        cursor (str | None): Cursor to pass as ``cursor`` to the next ``sync_resume`` call to resume
             from this position.
         has_next (bool): Whether more changes are available beyond this page.
         typing (TypeInformation | None): Property type information, present when the request was