# What Are CocoIndex Connectors? Architecture and Implementation Guide

> Discover CocoIndex connectors, modular bridges connecting CocoIndex to external systems via standardized APIs. Explore their architecture and implementation.

- Repository: [CocoIndex/cocoindex](https://github.com/cocoindex-io/cocoindex)
- Tags: how-to-guide
- Published: 2026-05-05

---

**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`](https://github.com/cocoindex-io/cocoindex/blob/main/python/cocoindex/connectors/localfs/_target.py), the provider is instantiated as follows:

```python
_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`](https://github.com/cocoindex-io/cocoindex/blob/main/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:

```python
@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`](https://github.com/cocoindex-io/cocoindex/blob/main/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:

```python
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:

```python
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:

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/kafka/_source.py) and [`qdrant/_target.py`](https://github.com/cocoindex-io/cocoindex/blob/main/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`](https://github.com/cocoindex-io/cocoindex/blob/main/_target.py) (e.g., [`python/cocoindex/connectors/postgres/_target.py`](https://github.com/cocoindex-io/cocoindex/blob/main/python/cocoindex/connectors/postgres/_target.py)), while source implementations use [`_source.py`](https://github.com/cocoindex-io/cocoindex/blob/main/_source.py) (e.g., [`python/cocoindex/connectors/kafka/_source.py`](https://github.com/cocoindex-io/cocoindex/blob/main/python/cocoindex/connectors/kafka/_source.py)).