connector-cursors

Connector Cursors - Typed Incremental Sync

Overview

Cursors enable incremental syncs by tracking progress between executions using typed Pydantic schemas. Instead of untyped dicts, each source defines a cursor schema with validated fields.

Architecture

Lifecycle

1. SyncFactory loads cursor from DB → instantiates typed Pydantic schema
2. Factory injects cursor into source via source.set_cursor()
3. Source reads cursor.data (dict), updates with cursor.update(**fields)
4. Orchestrator saves cursor.get() back to DB after sync

Key Components

• Cursor Schema (platform/cursors/{source}.py): Pydantic model defining cursor structure
• Runtime Cursor (SyncCursor): Wrapper providing update() and data API
• Source Decorator: cursor_class=MyCursor links source to schema (direct class reference, not string!)

Creating a Cursor Schema

Create a Pydantic model in platform/cursors/{source_name}.py:

from pydantic import Field
from ._base import BaseCursor

class GmailCursor(BaseCursor):
    """Gmail incremental sync cursor using history API."""
    history_id: str = Field(default="", description="Gmail history ID")

Common patterns:

Single field (API token):

class GoogleDriveCursor(BaseCursor):
    start_page_token: str = Field(default="", description="Drive Changes API token")

Multiple fields (per-resource):

class AirtableCursor(BaseCursor):
    table_cursors: Dict[str, str] = Field(
        default_factory=dict,
        description="Per-table cursor values (table_id -> last_modified_time)"
    )

Complex composite (metadata + per-resource):

class OutlookMailCursor(BaseCursor):
    delta_link: Optional[str] = Field(None, description="Primary delta link")
    folder_delta_links: Dict[str, str] = Field(default_factory=dict)
    folder_names: Dict[str, str] = Field(default_factory=dict)

Using Cursors in Sources

1. Import and Register Cursor