Sync

async AsyncCogniteClient.data_modeling.records.sync(
stream_id: str,
*,
initialize_cursor: str | None = None,
cursor: str | None = None,
filter: Filter | None = None,
sources: Sequence[RecordSourceSelector] | None = None,
target_units: RecordTargetUnits | Sequence[RecordTargetUnit] | None = None,
chunk_size: int = 1000,
include_typing: bool = False,
) → AsyncIterator[SyncRecordList]

Sync records from a stream.

Iterate over the change feed (new, updated and deleted records), yielding one chunk of records per request until the feed is exhausted (has_next is False). Pass exactly one of initialize_cursor (to start from a relative time such as "7d-ago") or cursor (to resume from where a previous sync left off). Each yielded SyncRecordList carries the cursor to persist for resuming later; this is also why records are always yielded in chunks rather than one by one.

Warning

Every chunk is fetched with a separate API request, so a small chunk_size increases the number of requests (i.e. comes at a high performance penalty). Keep the default of 1000 (the API maximum) unless your per-chunk processing genuinely needs smaller batches.

Parameters:

  • stream_id (str) – External ID of the stream to sync.

  • initialize_cursor (str | None) – Where to start, as a relative duration like "7d-ago". Mutually exclusive with cursor.

  • cursor (str | None) – Resume from a cursor from a previously yielded chunk. Mutually exclusive with initialize_cursor.

  • filter (Filter | None) – Filter expression (see 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.

  • chunk_size (int) – Number of records per yielded chunk, between 1 and 1000. Defaults to 1000.

  • include_typing (bool) – If True, include property type information on each yielded list’s typing attribute.

Yields:

SyncRecordList – One chunk of change records, with cursor and has_next set.

Examples

Iterate over all changes from the last 7 days, persisting the cursor after each processed chunk:

>>> from cognite.client import CogniteClient
>>> client = CogniteClient()
>>> for chunk in client.data_modeling.records.sync(
...     stream_id="my-stream", initialize_cursor="7d-ago"
... ):
...     for record in chunk:
...         pass  # process record; record.status is created/updated/deleted
...     last_cursor = chunk.cursor

Later, sync only what changed since then by resuming from the stored cursor:

>>> for chunk in client.data_modeling.records.sync(
...     stream_id="my-stream", cursor="previously-stored-cursor"
... ):
...     pass

Fetch chunks with manual control, e.g. to poll at your own cadence. Store the iterator in a variable to keep pulling chunks from where you left off:

>>> feed = client.data_modeling.records.sync(
...     stream_id="my-stream", cursor="previously-stored-cursor"
... )
>>> first_chunk = next(feed)
>>> second_chunk = next(feed)