What Are CocoIndex Connectors? Architecture and Implementation Guide

CocoIndex connectors are modular bridges that link CocoIndex’s declarative data-pipeline engine to external storage and compute systems through standardized Target and Source APIs.

CocoIndex connectors within the cocoindex-io/cocoindex repository enable pipelines to interact with external systems without imperative boilerplate. Each connector lives under python/cocoindex/connectors/ and consists of two complementary parts: a Target that declares and reconciles desired state in external systems, and a Source that provides streaming read access to external data. This dual architecture allows developers to mount PostgreSQL tables, local filesystem directories, Kafka streams, and vector databases using a consistent, declarative interface.

Connector Architecture: Targets and Sources

Every CocoIndex connector is split into two specialized modules that handle distinct I/O patterns.

• Target connectors (typically <system>/_target.py) declare what should exist in the external system—whether files, database rows, or vector embeddings. The engine reconciles these declarations against actual remote state, creating, updating, or deleting resources automatically.

• Source connectors (typically <system>/_source.py) provide a read API that yields records as a stream. These integrate with coco.mount_each() or coco.use_mount() to feed data into transformation pipelines.

Both components implement standardized interfaces, making it trivial to swap storage backends without modifying pipeline logic.

How Target Connectors Manage State

Target connectors implement a state reconciliation mechanism that guarantees idempotent, incremental updates to external systems.

Registration and State Providers

Each target registers a TargetStateProvider with the core engine. In python/cocoindex/connectors/localfs/_target.py, the provider is instantiated as follows:

_root_provider = coco.register_root_target_states_provider(...)

This provider translates stable keys (file paths, table names, or collection IDs) into concrete actions (_EntryAction, _RowAction, etc.) and maintains a fingerprint of the previous state for tracking.

Reconciliation Flow

When a pipeline component finishes, the engine invokes the provider’s reconcile method. The provider compares the desired specification (e.g., DirTarget.declare_file) against the previously recorded fingerprint. If differences exist, it emits an action (write file, insert row, delete document) executed by a sink via _action_sink_with_child.

Child Handling for Nested Resources

Targets containing nested structures (directories, database schemas) return a ChildTargetDef handler. This allows the engine to descend into sub-resources without losing identity, ensuring that parent containers are reconciled before child elements.

How Source Connectors Stream Data

Source connectors expose iterator classes that support both synchronous and asynchronous consumption patterns.

Row Fetching and Type Safety

In python/cocoindex/connectors/postgres/_source.py, the PgTableSource class returns a RowFetcher that manages connection pooling and query execution. The fetcher accepts an optional row_type parameter—typically a dataclass—that triggers a row factory transformation:

@dataclass
class Article:
    id: int
    title: str
    body: str

pg_source = postgres.PgTableSource(
    pool=await asyncpg.create_pool(dsn="postgresql://user@localhost/db"),
    table_name="articles",
    row_type=Article,  # Columns inferred from dataclass fields

)

Async Streaming Support

Sources like the Kafka connector (python/cocoindex/connectors/kafka/_source.py) implement KafkaMessageFetcher to consume streams asynchronously. These iterators internally manage connection lifecycle, applying backpressure and batching as configured, then yield raw records for transformation.

Implementing CocoIndex Connectors: Code Examples

Writing Files with the Local Filesystem Target

The following example demonstrates mounting a directory target and declaring files declaratively:

import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs

# Mount a directory target (creates the directory if missing)

target = coco.use_mount(
    coco.component_subpath("output"),
    localfs.declare_dir_target,
    pathlib.Path("./out"),
)

# Declare files inside the directory

target.declare_file("hello.txt", content="Hello, world!", create_parent_dirs=True)
target.declare_file("data.json", content=b'{"x": 1, "y": 2}')

Here, declare_dir_target creates a DirTarget instance. Under the hood, the provider registers actions via _action_sink_with_child, and upon component completion, the engine writes the files while removing any stale entries in ./out.

Reading from PostgreSQL and Writing to LanceDB

This pipeline extracts rows from PostgreSQL, computes embeddings, and stores vectors in LanceDB:

import cocoindex as coco
from cocoindex.connectors import postgres, lancedb
from dataclasses import dataclass

@dataclass
class Article:
    id: int
    title: str
    body: str

# Configure PostgreSQL source with typed rows

pg_source = postgres.PgTableSource(
    pool=await asyncpg.create_pool(dsn="postgresql://user@localhost/db"),
    table_name="articles",
    row_type=Article,
)

# Mount LanceDB target for vector storage

vector_target = coco.use_mount(
    coco.component_subpath("vectors"),
    lancedb.declare_dir_target,
    pathlib.Path("./vectors"),
)

@coco.fn(memo=True)
async def embed_and_store(row: Article, target: lancedb.DirTarget) -> None:
    embedding = await compute_embedding(row.body)
    target.declare_file(
        f"{row.id}.npy",
        content=embedding.tobytes(),
        create_parent_dirs=True,
    )

# Stream rows and process each

await coco.mount_each(
    embed_and_store,
    pg_source.fetch_rows().items(key=lambda r: r.id),
    vector_target,
)

The fetch_rows() method returns a RowFetcher that yields Article instances. The mount_each function pairs each row with the target, invoking embed_and_store to declare vector files. The engine reconciles the target directory, ensuring only new or changed embeddings are written.

Streaming from Kafka to Qdrant

For event streaming pipelines, combine a Kafka source with a Qdrant vector target:

from cocoindex.connectors import kafka, qdrant

kafka_source = kafka.KafkaSource(
    bootstrap_servers="localhost:9092",
    topic="events",
    value_type=bytes,
)

qdrant_target = coco.use_mount(
    coco.component_subpath("vectors"),
    qdrant.declare_collection_target,
    collection_name="event_vectors",
)

@coco.fn(memo=True)
async def ingest_message(msg: bytes, target: qdrant.CollectionTarget) -> None:
    vec = await encode_message(msg)
    target.upsert(ids=[msg_id], vectors=[vec])

await coco.mount_each(
    ingest_message,
    kafka_source.fetch_messages().items(key=lambda m: m.id),
    qdrant_target,
)

This pattern applies across connector implementations found in kafka/_source.py and qdrant/_target.py.

Summary

• CocoIndex connectors consist of Target modules for declarative writes and Source modules for streaming reads, both located under python/cocoindex/connectors/.
• Target connectors use TargetStateProvider registration and reconcile methods to compare desired state against remote fingerprints, emitting actions like _EntryAction or _RowAction via _action_sink_with_child.
• Source connectors expose fetcher classes (e.g., RowFetcher, KafkaMessageFetcher) that support typed deserialization through row factories and work with both sync and async iteration.
• The child handling mechanism via ChildTargetDef enables reconciliation of nested resources such as directories or table schemas.
• Connectors enable idempotent, incremental pipelines that automatically create, update, or delete external resources to match declared specifications.

Frequently Asked Questions

What is the difference between a Target and a Source in CocoIndex?

A Target handles write operations by declaring desired state in external systems like PostgreSQL, LanceDB, or local filesystems; it reconciles these declarations against actual remote state to determine necessary create, update, or delete actions. A Source handles read operations, providing streaming access to external data through fetcher classes like RowFetcher or KafkaMessageFetcher that integrate with coco.mount_each().

How does CocoIndex ensure idempotent writes to external systems?

Idempotency is achieved through the TargetStateProvider reconciliation loop. The provider stores a fingerprint of previously written state alongside stable keys (file paths, row IDs). When the pipeline runs, the provider compares the current desired specification against this fingerprint; only when differences are detected does it emit actions to modify the external system, preventing redundant writes.

Can I use custom data types when reading from PostgreSQL sources?

Yes. The PgTableSource class accepts a row_type parameter, typically a Python dataclass, which triggers an internal row factory that maps PostgreSQL columns to dataclass fields. This provides type-safe access to database records within pipeline functions without manual dictionary parsing.

Where are connector implementations located in the repository?

All connectors reside under python/cocoindex/connectors/, with each subdirectory representing a specific external system. Target implementations are conventionally named _target.py (e.g., python/cocoindex/connectors/postgres/_target.py), while source implementations use _source.py (e.g., python/cocoindex/connectors/kafka/_source.py).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}