Target Connector

Target Connector

Overview

A target connector connects CocoIndex's declarative target state system to external systems. It handles synchronization by determining what changed and applying changes to the external system.

When to Use

Use this skill when creating a new target connector for any external system (databases, file systems, cloud storage, APIs, etc.).

Key Data Types

What You Implement

What CocoIndex Provides

Implementation Workflow

Root Target States

1. Define types: Key, Spec, TrackingRecord, Action 2. Implement TargetHandler: The reconcile() method must be non-blocking 3. Create TargetActionSink: Use TargetActionSink.from_fn() or from_async_fn(). The callback receives context_provider: ContextProvider as its first positional argument, followed by actions 4. Register provider: Call register_root_target_states_provider(name, handler) 5. Create user-facing API: Wrap the provider in a user-friendly class

Non-Root (Child) Target States

For targets nested inside another target (e.g., files inside a directory):

1. Parent sink returns ChildTargetDef(handler=...) when executed 2. Call declare_target_state_with_child(parent_ts) to get an unresolved child provider 3. CocoIndex resolves the child provider when parent's sink executes

Child Invalidation

For container targets, set child_invalidation in TargetReconcileOutput when a container change affects its children:

Pattern for two-level (table/row) connectors using `statediff.diff_composite`:

# After computing main_action and column_actions via statediff.diff_composite:
child_invalidation: Literal["destructive", "lossy"] | None = None
if main_action == "replace":
    # Table dropped and recreated — all rows are destroyed.
    child_invalidation = "destructive"
elif main_action is None and any(a != "insert" for a in column_actions.values()):
    # Column changes other than adding new columns may lose existing row data.
    child_invalidation = "lossy"

return coco.TargetReconcileOutput(
    action=_TableAction(...),
    sink=self._sink,
    tracking_record=_TableTrackingRecord(...),
    child_invalidation=child_invalidation,
)

For connectors without column-level diffs (e.g., a collection that is either intact or fully replaced), only "destructive" applies:

child_invalidation: Literal["destructive"] | None = (
    "destructive" if main_action == "replace" else None
)

TargetHandler Protocol

class TargetHandler(Protocol[ValueT, TrackingRecordT, OptChildHandlerT]):
    def reconcile(
        self,
        key: StableKey,
        desired_target_state: ValueT | NonExistenceType,
        prev_possible_records: Collection[TrackingRecordT],
        prev_may_be_missing: bool,
        /,
    ) -> TargetReconcileOutput[Any, TrackingRecordT, OptChildHandlerT] | None:
        ...

    # Optional: override to support attachment types
    def attachment(self, att_type: str) -> TargetHandler | None:
        return None

Parameters:

• key: StableKey — a union of None | bool | int | str | bytes | uuid.UUID | Symbol | tuple[StableKey, ...]
• desired_target_state: What the user declared, or NON_EXISTENCE if no longer declared
• prev_possible_records: Tracking records from previous runs (may have multiple)
• prev_may_be_missing: If True, the target state might not exist in the external system

Returns:

• TargetReconcileOutput(action, sink, tracking_record, child_invalidation=None) if an action is needed (generic params: [ActionT, TrackingRecordT, OptChildHandlerT])
• None if no changes are required

The optional child_invalidation field is only relevant for container targets — see Child Invalidation.

Important: The reconcile() method must be non-blocking. It should only compare states and return an action — actual I/O happens in the sink.

Best Practices

Use ContextKey for External Resource Identity

When a target connector manages state in an external resource (database, object store, etc.), use a ContextKey string as part of the target state key — not connection parameters like host, port, or credentials.

Why: Target state keys must be stable across runs for correct reconciliation. CocoIndex uses keys to match current declarations with previously tracked states. If the key is stable, previously tracked states are associated with the current target, so CocoIndex can correctly reconcile — e.g., deleting rows that are no longer declared. If the key changes (because a connection parameter changed), CocoIndex cannot associate previous tracked states with the current target, and treats the target as being in a cleared state — losing the ability to clean up old data.

Pattern:

# User creates a stable logical name for the resource
db = coco.ContextKey[asyncpg.Pool]("my_pg")

# Target connector uses db.key (the string "my_pg") in the target state key
class _TableKey(NamedTuple):
    db_key: str           # Stable — from ContextKey.key
    schema_name: str | None
    table_name: str

key = _TableKey(db_key=db.key, ...)

# At action time, resolve the live connection from context_provider
pool = context_provider.get(key.db_key, asyncpg.Pool)

This decouples target identity from transient connection details — changing a password, switching replicas, or rotating credentials won't invalidate tracked states.

Reference: See _TableKey in python/cocoindex/connectors/postgres/_target.py and python/cocoindex/connectors/surrealdb/_target.py.

Idempotent Actions

Actions should be idempotent:

# Good
path.mkdir(parents=True, exist_ok=True)
path.unlink(missing_ok=True)
await conn.execute("INSERT ... ON CONFLICT DO UPDATE ...")

# Bad
path.mkdir()  # Fails if exists
await conn.execute("INSERT ...")  # Fails on duplicate key

Handle Multiple Previous States

Due to interrupted updates, prev_possible_records may contain multiple records:

if not prev_may_be_missing and all(
    prev.fingerprint == target_fp for prev in prev_possible_records
):
    return None  # Safe to skip

Fingerprinting for Change Detection

Use the connectorkits.fingerprint utilities for content-based change detection:

from cocoindex.connectorkits.fingerprint import fingerprint_bytes, fingerprint_str, fingerprint_object

# For raw bytes
fp = fingerprint_bytes(content)

# For strings
fp = fingerprint_str(text)

# For arbitrary objects (uses memo key mechanism)
fp = fingerprint_object(obj)

Shared Action Sinks

Create module-level shared sinks when all handler instances use the same action logic. The callback must accept context_provider: ContextProvider as its first positional argument:

def _apply_actions(
    context_provider: ContextProvider, actions: Sequence[MyAction]
) -> list[coco.ChildTargetDef[MyChildHandler] | None] | None:
    for action in actions:
        conn = context_provider.get(action.key.db_key, ConnType)
        ...

_shared_sink = coco.TargetActionSink.from_fn(_apply_actions)

Input Safety

When building queries from user-provided names (table, column, index) or values (record IDs, keys), you must guard against injection and ensure correctness. See input_safety.md for patterns on identifier validation, parameterized queries, and value escaping.

Completion Checklist

After implementing the connector code, complete these additional steps:

1. Optional Dependencies

If the connector requires third-party packages, update pyproject.toml:

[project.optional-dependencies]
# Add new optional dependency group
myconnector = ["some-package>=1.0.0"]

# Add to the 'all' group
all = [
    # ... existing deps ...
    "some-package>=1.0.0",
]

[[tool.mypy.overrides]]
# Add to mypy ignore list if package lacks type stubs
module = [
    # ... existing modules ...
    "some_package",
    "some_package.*",
]
ignore_missing_imports = true

2. Documentation

Create connector documentation at docs/docs/connectors/<connector_name>.md:

• Follow the structure of existing connector docs (e.g., postgres.md, sqlite.md)
• Include: connection setup, target state APIs, schema definition, type mappings, examples
• Add a note about optional dependencies if applicable

Update docs/sidebars.ts to include the new connector:

{
  type: 'category',
  label: 'Connectors',
  items: [
    // ... existing connectors ...
    'connectors/<connector_name>',  // Add in alphabetical order
  ],
},

3. Tests

Create tests at python/tests/connectors/test_<connector_name>_target.py:

Test structure:

import pytest
import cocoindex as coco
from tests import common

# Check for optional dependency availability
try:
    import optional_package
    HAS_OPTIONAL = True
except ImportError:
    HAS_OPTIONAL = False

requires_optional = pytest.mark.skipif(
    not HAS_OPTIONAL, reason="optional-package is not installed"
)

coco_env = common.create_test_env(__file__)

Required test cases:

Test pattern:

DB_KEY = coco.ContextKey[connector.ConnectionType]("test_db")

def test_insert_and_update(connector_fixture: tuple[Connection, Path]) -> None:
    conn, _ = connector_fixture
    source_rows: list[RowType] = []

    coco_env.context_provider.provide(DB_KEY, conn)

    async def declare_target() -> None:
        table = await coco.use_mount(
            coco.component_subpath("setup", "table"),
            connector.declare_table_target,
            DB_KEY,
            "test_table",
            await connector.TableSchema.from_class(RowType, primary_key=["id"]),
        )
        for row in source_rows:
            table.declare_row(row=row)

    app = coco.App(
        coco.AppConfig(name="test_insert", environment=coco_env),
        declare_target,
    )

    # Insert
    source_rows.append(RowType(id="1", name="Alice"))
    app.update()
    assert read_data(conn, "test_table") == [{"id": "1", "name": "Alice"}]

    # Update
    source_rows[0] = RowType(id="1", name="Alice Updated")
    app.update()
    assert read_data(conn, "test_table") == [{"id": "1", "name": "Alice Updated"}]

Optional feature tests:

@requires_optional
def test_vector_support(connector_with_vec: tuple[Connection, Path]) -> None:
    """Tests that require optional dependencies should be skipped when unavailable."""
    # ... test vector functionality ...

Reference implementations:

• python/tests/connectors/test_sqlite_target.py - SQLite tests with vector support

Attachment Providers

For targets with auxiliary child states (e.g., indexes on a database table), see attachments.md for the full reference on implementing attachment providers.

Resources

For complete implementation details and examples, see:

• docs/docs/advanced_topics/custom_target_connector.md - Full documentation
• python/cocoindex/connectors/localfs/_target.py - File system target connector (sync API, nested directory targets)
• python/cocoindex/connectors/sqlite/_target.py - SQLite target connector (sync API, two-level table/row targets, vector support)
• python/cocoindex/connectors/postgres/_target.py - PostgreSQL target connector (async API, two-level table/row targets, vector support, attachment providers)
• python/cocoindex/connectors/doris/_target.py - Doris target connector (async API, two-level table/row targets, Stream Load bulk inserts)

Attachment Providers

Overview

Attachment providers allow a target handler to expose auxiliary child target states that coexist with regular children under the same parent. For example, a database table handler manages rows as regular children, but also supports vector indexes and SQL command attachments as auxiliary states.

Attachment providers use symbol keys (prefixed with @) to namespace-separate them from regular children, avoiding path conflicts.

When to Use

Use attachment providers when a target has auxiliary state beyond its primary children:

• Database indexes (vector indexes, B-tree indexes)
• SQL commands (triggers, materialized views)
• Any metadata or configuration that lives alongside the primary data

Path Hierarchy

Attachment target states live under symbol-keyed sub-providers within the same parent:

table "my_table"               (root target state — table)
├── row "id=1"                  (regular child — row)
├── row "id=2"                  (regular child — row)
├── @vector_index               (attachment namespace)
│   └── "embedding_idx"         (attachment target state — vector index)
└── @sql_command_attachment      (attachment namespace)
    └── "custom_idx"            (attachment target state — SQL command)

The @ prefix is a symbol key that separates attachment namespaces from regular child keys.

How It Works

1. Parent handler implements attachment(att_type) returning a handler for that attachment type (or None if unsupported) 2. User code calls provider.attachment(att_type) on a resolved child provider to get an attachment sub-provider 3. Target states declared under the attachment provider are tracked independently from regular children 4. Attachment providers are cached — calling .attachment("x") twice returns the same provider

Implementation

Step 1: Define Attachment Types

Define spec, action, and tracking record types for each attachment kind:

class _VectorIndexSpec(NamedTuple):
    column: str
    metric: str
    method: str
    lists: int | None
    m: int | None
    ef_construction: int | None

Step 2: Implement Attachment Handler

Create a handler class with reconcile() method:

class _VectorIndexHandler:
    def __init__(self, pool, table_name, schema_name):
        self._pool = pool
        self._table_name = table_name
        self._schema_name = schema_name
        self._sink = coco.TargetActionSink.from_async_fn(self._apply_actions)

    def reconcile(
        self,
        key: coco.StableKey,
        desired_state: _VectorIndexSpec | coco.NonExistenceType,
        prev_possible_records: Collection[_VectorIndexFingerprint],
        prev_may_be_missing: bool,
        /,
    ) -> coco.TargetReconcileOutput[_VectorIndexAction, _VectorIndexFingerprint] | None:
        # Compare desired state with previous, return action or None
        ...

Step 3: Add attachment() to Parent Handler

The parent handler (e.g., _RowHandler) returns the appropriate attachment handler:

class _RowHandler:
    def attachment(self, att_type: str) -> _VectorIndexHandler | _SqlCommandHandler | None:
        if att_type == "vector_index":
            return _VectorIndexHandler(self._pool, self._table_name, self._schema_name)
        elif att_type == "sql_command_attachment":
            return _SqlCommandHandler(self._pool, self._table_name, self._schema_name)
        return None

Step 4: Expose User-Facing API

Wrap the attachment provider in a convenient method on the target class:

class TableTarget:
    def declare_vector_index(self, *, name, column, metric="cosine", method="ivfflat", ...):
        spec = _VectorIndexSpec(column=column, metric=metric, method=method, ...)
        att_provider = self._provider.attachment("vector_index")
        coco.declare_target_state(att_provider.target_state(name, spec))

    def declare_sql_command_attachment(self, *, name, setup_sql, teardown_sql=None):
        spec = _SqlCommandSpec(setup_sql=setup_sql, teardown_sql=teardown_sql)
        att_provider = self._provider.attachment("sql_command_attachment")
        coco.declare_target_state(att_provider.target_state(name, spec))

Tracking Record Design

Choose the tracking record type based on whether teardown recovery is needed:

Fingerprint example (vector index): Only needs to detect whether the spec changed. On change or delete, the action is always DROP + CREATE — no previous state info needed.

tracking_record = fingerprint_object(desired_state)  # bytes

Full spec example (SQL command): On change or delete, the previous teardown_sql must be executed before the new setup_sql. The full spec is stored so prev_possible_records contains recoverable teardown information.

tracking_record = desired_state  # _SqlCommandSpec (the spec itself)

Reference Implementations

• python/cocoindex/connectors/postgres/_target.py — _VectorIndexHandler, _SqlCommandHandler, _RowHandler.attachment(), TableTarget.declare_vector_index(), TableTarget.declare_sql_command_attachment()

Input Safety: Identifiers & Values

Target connectors build queries from user-provided names and values. Follow these guidelines to prevent injection and ensure correctness.

1. Identifier Validation

User-provided names (table, column, index) are interpolated into queries as identifiers. Validate them early at API entry points — not at query construction time.

Use a regex allowlist to reject anything that isn't a simple identifier:

import re

_IDENTIFIER_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")

def _validate_identifier(name: str, kind: str) -> None:
    """Raise ValueError if name is not a safe identifier."""
    if not _IDENTIFIER_RE.match(name):
        raise ValueError(
            f"Invalid {kind}: {name!r}. "
            f"Must match [a-zA-Z_][a-zA-Z0-9_]*."
        )

Call at every public method that accepts a name:

def table_target(self, table_name: str, ...) -> ...:
    _validate_identifier(table_name, "table name")
    ...

class TableSchema:
    def __init__(self, columns: dict[str, ColumnDef], ...) -> None:
        for col_name in columns:
            _validate_identifier(col_name, "column name")
        ...

2. Parameterized Queries for Values

Always use parameterized queries (bind variables) for data values. Never interpolate values directly into query strings.

# Good — parameterized
await conn.execute("INSERT INTO t (name) VALUES ($1)", value)       # PostgreSQL
conn.execute("INSERT INTO t (name) VALUES (?)", (value,))           # SQLite
await conn.query("UPSERT t:id CONTENT $content", {"content": val})  # SurrealDB

# Bad — string interpolation
await conn.execute(f"INSERT INTO t (name) VALUES ('{value}')")

3. Value Escaping (When Parameterization Isn't Possible)

Some query languages require inline values in certain positions (e.g., SurrealDB record IDs in table:id syntax). In these cases:

• Preserve type distinctions. Integer 123 and string "123" may be semantically different (e.g., table:123 vs ` table:123 ` in SurrealDB).
• Escape the quoting character. If the target uses backtick quoting, escape backslashes and backticks inside the value.

def _format_record_id(value: Any) -> str:
    """Format a record ID for inline use, preserving type."""
    if isinstance(value, (int, float)):
        return str(value)                              # bare numeric: 123, 3.14
    s = str(value)
    s = s.replace("\\", "\\\\").replace("`", "\\`")
    return f"`{s}`"                                    # quoted string: `alice`

4. Testing

Add tests for safety helpers that don't require a database:

• Valid identifiers pass, invalid ones raise ValueError
• Escaping produces correct output for special characters, empty strings, numeric types
• API entry points reject bad names (e.g., TableSchema(columns={"bad-name": ...}))

Add integration tests for round-tripping values with special characters through the full upsert/select cycle when a database is available.

Reference: See python/cocoindex/connectors/surrealdb/_target.py and python/tests/connectors/test_surrealdb_target.py for the canonical implementation.