feat(records): add sync endpoint

Add RecordsAPI.sync, a cursor-based POST to /streams/{streamId}/records/sync
returning one page as a SyncRecordList that carries the resumable `cursor`,
`has_next`, and optional `typing`. Mirrors how instances.sync surfaces a
resumable cursor (the caller persists `.cursor` and passes it back) rather than
draining and discarding it.

- SyncRecord extends Record with a `status` (created/updated/deleted); deleted
  tombstones load with properties=None.
- Provide exactly one of `cursor` or `initialize_cursor` ("<duration>-ago");
  passing both raises ValueError.
- Reuses the records read model + RETRIEVE read semaphore from the list PR.

Stacked on feat/records-list. target_units deferred to a follow-up.

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

Author: andersfylling

cognite/client/_api/data_modeling/records.py

@@ -12,6 +12,7 @@
     RecordList,
     RecordSourceSelector,
     RecordWrite,
+    SyncRecordList,
     TimeRange,
 )
 from cognite.client.data_classes.filters import Filter
@@ -205,3 +206,75 @@ async def list(
             semaphore=self._get_semaphore("read"),
         )
         return RecordList._load_raw_api_response([response.json()])
+
+    async def sync(
+        self,
+        stream_id: str,
+        *,
+        cursor: str | None = None,
+        initialize_cursor: str | None = None,
+        filter: Filter | None = None,
+        sources: Sequence[RecordSourceSelector] | None = None,
+        limit: int | None = None,
+        include_typing: bool = False,
+    ) -> 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.
+
+        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.
+            filter (Filter | None): Filter expression (see :mod:`cognite.client.data_classes.filters`).
+            sources (Sequence[RecordSourceSelector] | None): Which container properties to return.
+            limit (int | None): Maximum number of records to return in this page (1-1000).
+            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.
+
+        Examples:
+
+            Initialize a sync, process the page, then resume from the cursor later:
+
+                >>> from cognite.client import CogniteClient
+                >>> client = CogniteClient()
+                >>> page = client.data_modeling.records.sync(
+                ...     stream_id="my-stream", initialize_cursor="7d-ago"
+                ... )
+                >>> for record in page:
+                ...     pass  # process record; record.status is created/updated/deleted
+                >>> next_page = client.data_modeling.records.sync(
+                ...     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.")
+        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()
+        if sources is not None:
+            body["sources"] = [source.dump() for source in sources]
+        if limit is not None:
+            body["limit"] = limit
+        if include_typing:
+            body["includeTyping"] = True
+
+        response = await self._post(
+            url_path=self._records_url(stream_id, "/sync"),
+            json=body,
+            semaphore=self._get_semaphore("read"),
+        )
+        return SyncRecordList._load_response(response.json())

cognite/client/_sync_api/data_modeling/records.py

@@ -128,6 +128,8 @@
     RecordSourceSelector,
     RecordWrite,
     RecordWriteList,
+    SyncRecord,
+    SyncRecordList,
     TimeRange,
 )
 from cognite.client.data_classes.data_modeling.spaces import Space, SpaceApply, SpaceApplyList, SpaceList
@@ -285,6 +287,8 @@
     "StreamTemplateWriteSettings",
     "StreamWrite",
     "SubscriptionContext",
+    "SyncRecord",
+    "SyncRecordList",
     "Text",
     "TimeRange",
     "TimeSeriesReference",

cognite/client/data_classes/data_modeling/__init__.py

@@ -2,7 +2,7 @@
 
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Literal
 
 from typing_extensions import Self
 
@@ -26,6 +26,8 @@
     "RecordSourceSelector",
     "RecordWrite",
     "RecordWriteList",
+    "SyncRecord",
+    "SyncRecordList",
     "TimeRange",
 ]
 
@@ -270,3 +272,90 @@ def _load(cls, resource: dict[str, Any]) -> Self:
 
     def dump(self, camel_case: bool = True) -> dict[str, Any]:
         return {"source": self.source.dump(camel_case=camel_case), "properties": self.properties}
+
+
+class SyncRecord(Record):
+    """A record returned by the sync endpoint, annotated with a change status.
+
+    For ``status="deleted"`` tombstones (mutable streams), :attr:`properties` is ``None``.
+
+    Args:
+        space (str): Space the record belongs to.
+        external_id (str): External ID of the record.
+        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
+            deleted tombstones).
+    """
+
+    def __init__(
+        self,
+        space: str,
+        external_id: str,
+        created_time: int,
+        last_updated_time: int,
+        status: Literal["created", "updated", "deleted"],
+        properties: dict[str, dict[str, dict[str, Any]]] | None = None,
+    ) -> None:
+        super().__init__(
+            space=space,
+            external_id=external_id,
+            created_time=created_time,
+            last_updated_time=last_updated_time,
+            properties=properties,
+        )
+        self.status = status
+
+    @classmethod
+    def _load(cls, resource: dict[str, Any]) -> Self:
+        return cls(
+            space=resource["space"],
+            external_id=resource["externalId"],
+            created_time=resource["createdTime"],
+            last_updated_time=resource["lastUpdatedTime"],
+            status=resource["status"],
+            properties=resource.get("properties"),
+        )
+
+    def dump(self, camel_case: bool = True) -> dict[str, Any]:
+        output = super().dump(camel_case=camel_case)
+        output["status"] = self.status
+        return output
+
+
+class SyncRecordList(CogniteResourceList[SyncRecord]):
+    """A page of :class:`SyncRecord` objects from the sync endpoint.
+
+    Args:
+        resources (Sequence[SyncRecord]): The records in this page.
+        cursor (str | None): Cursor to pass as ``cursor`` to the next ``sync`` 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
+            made with ``include_typing=True``.
+    """
+
+    _RESOURCE = SyncRecord
+
+    def __init__(
+        self,
+        resources: Sequence[SyncRecord],
+        cursor: str | None = None,
+        has_next: bool = False,
+        typing: TypeInformation | None = None,
+    ) -> None:
+        super().__init__(resources)
+        self.cursor = cursor
+        self.has_next = has_next
+        self.typing = typing
+
+    @classmethod
+    def _load_response(cls, response: dict[str, Any]) -> Self:
+        typing = TypeInformation._load(response["typing"]) if "typing" in response else None
+        return cls(
+            [SyncRecord._load(item) for item in response["items"]],
+            cursor=response["nextCursor"],
+            has_next=response["hasNext"],
+            typing=typing,
+        )