Cocoindex

CocoIndex

CocoIndex is a Python library for building incremental data processing pipelines with declarative target states. Think spreadsheets or React for data pipelines: declare what the output should look like based on current input, and CocoIndex automatically handles incremental updates, change detection, and syncing to external systems.

Overview

CocoIndex enables building data pipelines that:

• Automatically handle incremental updates: Only reprocess changed data
• Use declarative target states: Declare what should exist, not how to update
• Support any Python types: No custom DSL -- use dataclasses, Pydantic, NamedTuple
• Provide function memoization: Skip expensive operations when inputs/code unchanged
• Sync to multiple targets: PostgreSQL, SQLite, LanceDB, Qdrant, SurrealDB, Apache Doris, file systems, Kafka

Key principle: TargetState = Transform(SourceState)

When to Use This Skill

Use this skill when building pipelines that involve:

• Document processing: PDF/Markdown conversion, text extraction, chunking
• Vector embeddings: Embedding documents/code for semantic search
• Database transformations: ETL from source DB to target DB
• Knowledge graphs: Extract entities and relationships from data
• LLM-based extraction: Structured data extraction using LLMs
• File-based pipelines: Transform files from one format to another
• Incremental indexing: Keep search indexes up-to-date with source changes
• Streaming pipelines: Kafka-based real-time data processing

Quick Start: Creating a New Project

Initialize Project

cocoindex init my-project
cd my-project

This creates: main.py, pyproject.toml, .env, README.md.

Add Dependencies

# For vector embeddings with PostgreSQL
dependencies = ["cocoindex>=1.0.0", "sentence-transformers", "asyncpg"]

# For LLM extraction
dependencies = ["cocoindex>=1.0.0", "litellm", "instructor", "pydantic>=2.0"]

See references/setup_project.md for complete examples.

Run the Pipeline

pip install -e .
cocoindex update main.py

Core Concepts

1. Apps

An App is the top-level executable that binds a main function with parameters:

import cocoindex as coco

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    ...

app = coco.App(
    coco.AppConfig(name="MyApp"),
    app_main,
    sourcedir=pathlib.Path("./data"),
)

2. Functions (@coco.fn)

The @coco.fn decorator marks functions as CocoIndex processing functions. Add memo=True to skip re-execution when inputs/code are unchanged:

@coco.fn(memo=True)
async def expensive_operation(data: str) -> Result:
    # LLM call, embedding generation, heavy computation
    return await expensive_transform(data)

Key parameters:

• memo=True -- Enable memoization (skip if inputs/code unchanged)
• version=1 -- Explicit version bump to force re-execution
• batching=True -- Auto-batch concurrent calls (async only)
• runner=coco.GPU -- Serialize GPU-bound execution

3. Processing Components

A processing component groups an item's processing with its target states.

Mount components with mount_each() (preferred for lists) or mount():

# One component per item (preferred for lists)
await coco.mount_each(process_file, files.items(), target_table)

# Single component (subpath auto-derived from fn.__name__)
await coco.mount(setup_fn, arg1)

# Dependent component (blocks until result returned)
result = await coco.use_mount(init_fn)

# Explicit subpath (when you need a specific path, e.g. in loops)
await coco.mount(coco.component_subpath("item", item_id), process_item, item)

Key points:

• All mount APIs are async
• mount(), use_mount(), and mount_each() auto-derive subpath from fn.__name__; optional explicit subpath as first arg
• Use use_mount() when you need the return value
• Use stable component paths for proper memoization and cleanup

4. Target States

Declare what should exist -- CocoIndex handles creation/update/deletion:

# Database row target
table.declare_row(row=MyRecord(id=1, name="example"))

# File target
localfs.declare_file(outdir / "output.txt", content, create_parent_dirs=True)

# Kafka message target
topic_target.declare_target_state(key="msg-1", value=json.dumps(data))

5. Context for Shared Resources

Use ContextKey to share expensive resources (DB connections, models) across components:

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    builder.provide(EMBEDDER, SentenceTransformerEmbedder("all-MiniLM-L6-v2"))
    yield

# In processing functions:
embedder = coco.use_context(EMBEDDER)

The @coco.lifespan decorator registers the function to the default CocoIndex environment, which is shared among all apps by default. ContextKey also serves as the stable identity for sources/targets -- the key string must remain stable across runs.

6. ID Generation

Generate stable, unique identifiers that persist across incremental updates:

from cocoindex.resources.id import generate_id, IdGenerator

# Deterministic: same dep -> same ID
chunk_id = await generate_id(chunk.content)

# Always distinct: each call -> new ID, even with same dep
id_gen = IdGenerator()
for chunk in chunks:
    chunk_id = await id_gen.next_id(chunk.content)

7. Catch-Up vs Live Mode

By default, app.update() runs in catch-up mode: it scans all sources, processes what changed since the last run (memoized components are skipped), syncs target states, and returns. Each call still has to scan sources to discover changes.

Live mode keeps the app running after catch-up and lets components stream changes continuously from their sources (e.g., file watcher, Kafka consumer), applying them with very low latency.

# Enable live mode
app.update_blocking(live=True)
# Or: cocoindex update main.py -L

Two things are needed: (1) enable live mode on the app, and (2) use a source that supports live updates.

• `LiveMapView` sources (e.g., localfs.walk_dir(..., live=True)) scan current state first, then watch for changes. They also work in catch-up mode -- write the pipeline once, choose mode at run time.
• `LiveMapFeed` sources (e.g., kafka.topic_as_map()) only stream changes with no initial snapshot. mount_each() auto-detects these and creates a live component internally.

# LocalFS with live watching
files = localfs.walk_dir(sourcedir, live=True, ...)
await coco.mount_each(process_file, files.items(), target)

# Kafka -- inherently live
items = kafka.topic_as_map(consumer, ["my-topic"])
await coco.mount_each(process_message, items, target)

Common Pipeline Patterns

Pattern 1: File Transformation

import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs
from cocoindex.resources.file import FileLike, PatternFilePathMatcher

@coco.fn(memo=True)
async def process_file(file: FileLike, outdir: pathlib.Path) -> None:
    content = await file.read_text()
    transformed = transform_content(content)
    outname = file.file_path.path.stem + ".out"
    localfs.declare_file(outdir / outname, transformed, create_parent_dirs=True)

@coco.fn
async def app_main(sourcedir: pathlib.Path, outdir: pathlib.Path) -> None:
    files = localfs.walk_dir(
        sourcedir,
        recursive=True,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
    )
    await coco.mount_each(process_file, files.items(), outdir)

app = coco.App(
    coco.AppConfig(name="Transform"),
    app_main,
    sourcedir=pathlib.Path("./data"),
    outdir=pathlib.Path("./out"),
)

Pattern 2: Vector Embedding Pipeline

import pathlib
from dataclasses import dataclass
from typing import AsyncIterator, Annotated

import asyncpg
from numpy.typing import NDArray

import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
from cocoindex.resources.chunk import Chunk
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from cocoindex.resources.id import IdGenerator

DATABASE_URL = "postgres://cocoindex:cocoindex@localhost/cocoindex"
PG_DB = coco.ContextKey[asyncpg.Pool]("pg_db")
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

_splitter = RecursiveSplitter()

@dataclass
class DocEmbedding:
    id: int
    filename: str
    text: str
    embedding: Annotated[NDArray, EMBEDDER]  # Dimensions inferred from ContextKey
    chunk_start: int
    chunk_end: int

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with await asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        builder.provide(EMBEDDER, SentenceTransformerEmbedder("all-MiniLM-L6-v2"))
        yield

@coco.fn
async def process_chunk(
    chunk: Chunk, filename: pathlib.PurePath,
    id_gen: IdGenerator, table: postgres.TableTarget[DocEmbedding],
) -> None:
    table.declare_row(row=DocEmbedding(
        id=await id_gen.next_id(chunk.text),
        filename=str(filename),
        text=chunk.text,
        embedding=await coco.use_context(EMBEDDER).embed(chunk.text),
        chunk_start=chunk.start.char_offset,
        chunk_end=chunk.end.char_offset,
    ))

@coco.fn(memo=True)
async def process_file(file: FileLike, table: postgres.TableTarget[DocEmbedding]) -> None:
    text = await file.read_text()
    chunks = _splitter.split(text, chunk_size=2000, chunk_overlap=500)
    id_gen = IdGenerator()
    await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    target_table = await postgres.mount_table_target(
        PG_DB,
        table_name="embeddings",
        table_schema=await postgres.TableSchema.from_class(DocEmbedding, primary_key=["id"]),
    )
    target_table.declare_vector_index(column="embedding")

    files = localfs.walk_dir(sourcedir, recursive=True,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]))
    await coco.mount_each(process_file, files.items(), target_table)

app = coco.App(coco.AppConfig(name="Embedding"), app_main, sourcedir=pathlib.Path("./data"))

Pattern 3: LLM-Based Extraction

import instructor
from pydantic import BaseModel
from litellm import acompletion

_instructor_client = instructor.from_litellm(acompletion, mode=instructor.Mode.JSON)

class ExtractionResult(BaseModel):
    title: str
    topics: list[str]

@coco.fn(memo=True)  # Memo avoids re-calling LLM
async def extract_and_store(content: str, message_id: int, table) -> None:
    result = await _instructor_client.chat.completions.create(
        model="gpt-4",
        response_model=ExtractionResult,
        messages=[{"role": "user", "content": f"Extract topics: {content}"}],
    )
    table.declare_row(row=Message(id=message_id, title=result.title, content=content))

Connectors and Operations

CocoIndex provides connectors for reading from and writing to external systems:

For detailed connector documentation, see references/connectors.md.

Text and Embedding Operations

Text Splitting

from cocoindex.ops.text import RecursiveSplitter, detect_code_language

splitter = RecursiveSplitter()
language = detect_code_language(filename="example.py")
chunks = splitter.split(text, chunk_size=1000, chunk_overlap=200, language=language)

Embeddings

from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

embedder = SentenceTransformerEmbedder("sentence-transformers/all-MiniLM-L6-v2")
embedding = await embedder.embed(text)  # Returns NDArray

CLI Commands

cocoindex init my-project              # Create new project
cocoindex update main.py               # Run app
cocoindex update main.py:my_app        # Run specific app
cocoindex update main.py -L            # Run in live mode (continuous)
cocoindex update main.py --full-reprocess  # Reprocess everything
cocoindex drop main.py [-f]            # Drop and reset all state
cocoindex ls [main.py]                 # List apps
cocoindex show main.py [--tree]        # Show component paths

Best Practices

1. Use @coco.fn on All Processing Functions

Every function that participates in the pipeline (declares target states, calls mount APIs, etc.) must be decorated with @coco.fn.

2. Add Memoization for Expensive Operations

@coco.fn(memo=True)  # Skip re-execution when inputs/code unchanged
async def process_chunk(chunk, table):
    embedding = await embedder.embed(chunk.text)  # Expensive!
    table.declare_row(...)

3. Use Stable Component Paths

# Good: Stable identifiers
coco.component_subpath("file", str(file.file_path.path))
coco.component_subpath("record", record.id)

# Bad: Unstable identifiers
coco.component_subpath("file", file)      # Object reference
coco.component_subpath("idx", idx)        # Index changes

4. Use Context for Shared Resources

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with await asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        yield

5. Use Annotated[NDArray, CONTEXT_KEY] for Vectors

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

@dataclass
class Record:
    vector: Annotated[NDArray, EMBEDDER]  # Auto-infer dimensions from ContextKey

6. Use Convenience APIs for Targets

# Mount table target -- subpath is automatic
table = await postgres.mount_table_target(
    PG_DB,
    table_name="my_table",
    table_schema=await postgres.TableSchema.from_class(MyRecord, primary_key=["id"]),
)

Troubleshooting

Everything Reprocessing

Add memo=True to expensive functions:

@coco.fn(memo=True)  # Add this
async def process_item(item):
    ...

Memoization Not Working

Check component paths are stable. Use stable IDs, not object references.

Resources

references/

• [api_reference.md](references/api_reference.md): Quick API reference
• [connectors.md](references/connectors.md): Complete connector reference
• [patterns.md](references/patterns.md): Detailed pipeline patterns
• [setup_project.md](references/setup_project.md): Project setup guide
• [setup_database.md](references/setup_database.md): Database setup guide

Runnable examples

Every pattern above has a complete, runnable app under `examples/` — start from the one closest to the task and adapt it. Each has its own README.md and most Python examples have a .env.example; see `examples/AGENTS.md` for the full map, credentials, and per-example run commands. Good starting points:

• Vector search → text_embedding (Postgres), text_embedding_qdrant / _lancedb (other stores)
• Code search → code_embedding
• LLM extraction → hn_trending_topics, patient_intake_extraction_baml
• Knowledge graph → conversation_to_knowledge, meeting_notes_graph_neo4j
• Custom transform → files_transform, pdf_to_markdown

External

• CocoIndex Documentation — full text at llms-full.txt
• GitHub Examples

Version Note

This skill is for CocoIndex >=1.0.0 (v1). It uses a completely different API from v0.

v0 code is what you likely learned from training data — do not emit it. If you find yourself writing any of these symbols, you are using the removed v0 API:

When reading third-party tutorials or model memory that mention these v0 symbols, disregard them and use the patterns in this skill and references/api_reference.md instead.

CocoIndex API Reference

Quick reference for the most commonly used CocoIndex APIs.

Import Convention

import cocoindex as coco

All APIs live directly on the coco module. There is no separate cocoindex.asyncio module.

---

@coco.fn Decorator

Mark a function as a CocoIndex processing function.

@coco.fn
async def my_function(arg1: str) -> None: ...

@coco.fn(memo=True, version=1)
async def expensive_fn(data: str) -> Result: ...

# Force async interface for a sync function (useful for batching)
@coco.fn.as_async(memo=True, batching=True)
def batch_embed(texts: list[str]) -> list[NDArray]: ...

Parameters:

• memo: bool = False -- Enable memoization (skip if inputs/code unchanged)
• version: int | None = None -- Explicit version bump to force re-execution
• logic_tracking: "full" | "self" | None = "full" -- How code changes are tracked
• batching: bool = False -- Auto-batch concurrent calls (async only)
• max_batch_size: int | None = None -- Max batch size
• runner: Runner | None = None -- e.g. coco.GPU for serialized GPU execution

---

Mount APIs (all async)

All mount APIs accept an optional ComponentSubpath as their first argument. When omitted, the subpath is auto-derived from Symbol(fn.__name__). Provide an explicit subpath when mounting the same function multiple times, using multi-part paths, or needing a specific path name.

coco.mount()

Mount a processing component in the background.

# Subpath auto-derived from fn.__name__
handle = await coco.mount(processor_fn, *args, **kwargs)

# Explicit subpath
handle = await coco.mount(coco.component_subpath("name"), processor_fn, *args, **kwargs)

await handle.ready()  # Optional: wait until component finishes

Parameters:

• subpath (optional) -- Component subpath. Auto-derived from fn.__name__ when omitted.
• processor_fn -- Function (or LiveComponent class) to run.
• *args, **kwargs -- Arguments passed to the function.

Returns: ComponentMountHandle

coco.use_mount()

Mount a dependent component and return its result. Parent depends on the child.

# Subpath auto-derived from fn.__name__
result = await coco.use_mount(init_fn, *args, **kwargs)

# Explicit subpath
result = await coco.use_mount(coco.component_subpath("setup"), init_fn, *args, **kwargs)

Parameters:

• subpath (optional) -- Component subpath. Auto-derived from fn.__name__ when omitted.
• processor_fn -- Function to run.
• *args, **kwargs -- Arguments passed to the function.

Returns: The return value of processor_fn.

coco.mount_each()

Mount one component per item in a keyed iterable. Preferred for processing lists.

# Subpath auto-derived from fn.__name__
await coco.mount_each(process_file, files.items(), *extra_args)

# Explicit subpath
await coco.mount_each(coco.component_subpath("process"), process_file, files.items(), table)

Parameters:

• subpath (optional) -- Component subpath. Auto-derived from fn.__name__ when omitted.
• fn -- Function to run per item. Item value is passed as first argument.
• items -- Keyed iterable of (StableKey, T) pairs, or a LiveMapFeed for live mode.
• *args, **kwargs -- Additional arguments passed to fn after the item.

Returns: ComponentMountHandle

coco.mount_target()

Mount a target state, ensuring the container is applied before returning the child provider.

provider = await coco.mount_target(target_state)

Prefer connector convenience methods (postgres.mount_table_target(), etc.) which call this internally.

coco.map()

Run a function concurrently on each item. No processing components are created -- pure concurrent execution within the current component.

results = await coco.map(process_chunk, chunks, *extra_args)

Parameters:

• fn -- Async function to apply. Item is passed as first argument.
• items -- Iterable or async iterable.
• *args, **kwargs -- Additional arguments passed to fn after the item.

Returns: list[T]

coco.component_subpath()

Create a stable component path for mounting.

coco.component_subpath("setup")
coco.component_subpath("file", str(file_path))
coco.component_subpath("record", record.id)

# Chaining with /
subpath = coco.component_subpath("a") / "b" / "c"

# Context manager form (applies to all nested mount calls)
with coco.component_subpath("process"):
    for f in files:
        await coco.mount(coco.component_subpath(str(f.path)), process_file, f)

StableKey types: str | int | bool | bytes | uuid.UUID | Symbol | tuple[StableKey, ...]

---

Context System

coco.ContextKey[T]

Type-safe key for sharing resources. The key string is the stable identity across runs.

PG_DB = coco.ContextKey[asyncpg.Pool]("pg_db")
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

• detect_change=True -- Opt in to auto-invalidate dependent memos when value changes (models, configs)
• detect_change=False (default) -- For resources not affecting computation (DB connections, loggers)

builder.provide()

Register a resource in context (used in lifespan).

builder.provide(PG_DB, pool)
builder.provide_with(KEY, context_manager)         # Sync CM
await builder.provide_async_with(KEY, async_cm)    # Async CM

coco.use_context()

Retrieve a resource from context inside a processing function.

pool = coco.use_context(PG_DB)
embedder = coco.use_context(EMBEDDER)

---

Lifespan

@coco.lifespan

Define environment setup/teardown. Registered to the default environment.

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with await asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        builder.provide(EMBEDDER, SentenceTransformerEmbedder(MODEL))
        yield

Can also be sync:

@coco.lifespan
def coco_lifespan(builder: coco.EnvironmentBuilder) -> Iterator[None]:
    builder.settings.db_path = pathlib.Path("./my.db")
    yield

---

App

coco.App

app = coco.App(
    coco.AppConfig(name="MyApp"),
    main_fn,
    **params,
)

# Async
await app.update()
handle = app.update(live=True)  # Live mode
await app.drop()

# Sync (blocking)
app.update_blocking(report_to_stdout=True)
app.drop_blocking()

coco.AppConfig

coco.AppConfig(
    name="MyApp",                    # Required
    environment=env,                 # Optional: custom Environment
    max_inflight_components=1024,    # Optional: concurrency limit
)

Start/Stop (for programmatic usage)

# Context manager (preferred)
async with coco.runtime():
    await app.update()

# Or manually
await coco.start()
try:
    await app.update()
finally:
    await coco.stop()

# Sync variants
with coco.runtime():
    app.update_blocking()

---

Exception Handlers

Global (in lifespan)

builder.set_exception_handler(my_handler)

Scoped (in processing functions)

async with coco.exception_handler(my_handler):
    await coco.mount_each(process_file, files.items(), table)

Handler Signature

async def my_handler(exc: BaseException, ctx: coco.ExceptionContext) -> None:
    logger.error(f"Error in {ctx.stable_path}: {exc}")

ExceptionContext provides: env_name, stable_path, processor_name, mount_kind, parent_stable_path, is_background, source, original_exception.

---

Live Mode

Run the pipeline continuously, streaming changes from sources:

# CLI
cocoindex update main.py -L

# Programmatic
handle = app.update(live=True)
async for snapshot in handle.watch():
    print(snapshot.stats)

Sources that support live mode:

• localfs.walk_dir(..., live=True) -- File watching
• kafka.topic_as_map(consumer, topics) -- Kafka topic consumption

---

CLI Commands

cocoindex init [PROJECT_NAME]              # Create new project
cocoindex update APP_TARGET                # Run app once
cocoindex update APP_TARGET -L             # Run in live mode
cocoindex update APP_TARGET --full-reprocess  # Force reprocess all
cocoindex update APP_TARGET --reset        # Reset state before running
cocoindex drop APP_TARGET [-f]             # Drop app and all state
cocoindex ls [APP_TARGET] [--db PATH]      # List apps
cocoindex show APP_TARGET [--tree]         # Show component paths

APP_TARGET format: main.py, main.py:app_name, my_module:app_name

---

Text Operations

Import: from cocoindex.ops.text import ...

detect_code_language()

language = detect_code_language(filename="example.py")  # -> "python"

RecursiveSplitter

splitter = RecursiveSplitter()
chunks = splitter.split(
    text,
    chunk_size=1000,
    chunk_overlap=200,
    min_chunk_size=300,
    language="python",  # Syntax-aware splitting
)
# Each chunk: Chunk(text=..., start=TextPosition(...), end=TextPosition(...))

SeparatorSplitter

splitter = SeparatorSplitter(separators_regex=[r"\n\n", r"\n"])
chunks = splitter.split(text)

---

Embedding Operations

Import: from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

embedder = SentenceTransformerEmbedder("sentence-transformers/all-MiniLM-L6-v2")
embedding = await embedder.embed(text)  # -> NDArray (float32)

As VectorSchemaProvider:

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

@dataclass
class Record:
    vector: Annotated[NDArray, EMBEDDER]  # Auto-infer dimensions

---

File Resources

Import: from cocoindex.resources.file import ...

FileLike (async file object)

text = await file.read_text()
data = await file.read()  # bytes, lazy/cached
fp = await file.content_fingerprint()

PatternFilePathMatcher

matcher = PatternFilePathMatcher(
    included_patterns=["**/*.py", "**/*.md"],
    excluded_patterns=[".*/**", "__pycache__/**"],
)

---

ID Generation

Import: from cocoindex.resources.id import ...

# Deterministic: same dep -> same ID
chunk_id = await generate_id(chunk.content)
chunk_uuid = generate_uuid(chunk.content)

# Distinct per call (even with same dep)
id_gen = IdGenerator()
chunk_id = await id_gen.next_id(chunk.content)

uuid_gen = UuidGenerator()
chunk_uuid = uuid_gen.next_uuid(chunk.content)

---

Vector Schema

Import: from cocoindex.resources.schema import VectorSchema

# Via ContextKey (preferred -- auto-infer from embedder)
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

@dataclass
class Record:
    vector: Annotated[NDArray, EMBEDDER]

# Explicit dimensions
schema = VectorSchema(dtype=np.float32, size=384)

@dataclass
class Record:
    vector: Annotated[NDArray, schema]

---

See Also

• Connectors Reference -- Database and system connectors
• Patterns Reference -- Common pipeline patterns
• Setup Project -- Project setup guide
• Setup Database -- Database setup guide

CocoIndex Connectors Reference

Comprehensive reference for all CocoIndex connectors.

Common Patterns

Target Setup

All target connectors follow this pattern:

1. Provide connection via ContextKey in lifespan 2. Mount target using mount_*_target(ContextKey, ...) convenience method 3. Declare rows/points/files on the returned target object

Target connectors take a ContextKey (not the connection object directly) as their first argument. This enables CocoIndex to track the identity of the target across runs.

Vector Support

Most connectors support vector embeddings via VectorSchemaProvider:

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

@dataclass
class Record:
    vector: Annotated[NDArray, EMBEDDER]  # Auto-infer dimensions from ContextKey

---

PostgreSQL

Import: from cocoindex.connectors import postgres

Capabilities: Source + Target | Vector: pgvector

Connection Setup

import asyncpg
import cocoindex as coco
from cocoindex.connectors import postgres

PG_DB = coco.ContextKey[asyncpg.Pool]("pg_db")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with await asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        yield

As Target

from dataclasses import dataclass
from typing import Annotated
from numpy.typing import NDArray

@dataclass
class Embedding:
    id: int
    text: str
    vector: Annotated[NDArray, EMBEDDER]

# Mount table target
target_table = await postgres.mount_table_target(
    PG_DB,
    table_name="embeddings",
    table_schema=await postgres.TableSchema.from_class(Embedding, primary_key=["id"]),
    pg_schema_name="my_schema",  # Optional, default "public"
)

# Optional: declare vector index
target_table.declare_vector_index(column="vector")

# Declare rows
target_table.declare_row(row=Embedding(id=1, text="hello", vector=vec))

As Source

@dataclass
class Record:
    id: int
    name: str

source = postgres.PgTableSource(
    coco.use_context(PG_DB),
    table_name="my_table",
    row_type=Record,
)

# Iterate (async)
fetcher = source.fetch_rows()
async for record in fetcher:
    ...

# Keyed iteration for mount_each
await coco.mount_each(process_record, source.fetch_rows().items(key=lambda r: r.id), table)

Type Mapping

SQL Command Attachments

target_table.declare_sql_command_attachment(
    name="my_index",
    setup_sql="CREATE INDEX ...",
    teardown_sql="DROP INDEX ...",  # Optional
)

---

SQLite

Import: from cocoindex.connectors import sqlite

Capabilities: Target only | Vector: sqlite-vec

Connection Setup

from cocoindex.connectors import sqlite

SQLITE_DB = coco.ContextKey[sqlite.ManagedConnection]("sqlite_db")

@coco.lifespan
def coco_lifespan(builder: coco.EnvironmentBuilder) -> Iterator[None]:
    conn = sqlite.connect("./data.db", load_vec="auto")
    builder.provide(SQLITE_DB, conn)
    yield
    conn.close()

As Target

target_table = await sqlite.mount_table_target(
    SQLITE_DB,
    table_name="embeddings",
    table_schema=await sqlite.TableSchema.from_class(Embedding, primary_key=["id"]),
)

target_table.declare_row(row=Embedding(id=1, text="hello", vector=vec))

Type Mapping

---

LanceDB

Import: from cocoindex.connectors import lancedb

Capabilities: Target only | Vector: Native | Storage: Local or cloud (S3, GCS)

Connection Setup

from cocoindex.connectors import lancedb

LANCE_DB = coco.ContextKey[lancedb.LanceAsyncConnection]("lance_db")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    conn = await lancedb.connect_async("./lancedb_data")  # or "s3://bucket/path"
    builder.provide(LANCE_DB, conn)
    yield

As Target

target_table = await lancedb.mount_table_target(
    LANCE_DB,
    table_name="embeddings",
    table_schema=await lancedb.TableSchema.from_class(Embedding, primary_key=["id"]),
)

target_table.declare_row(row=Embedding(id=1, text="hello", vector=vec))

Type Mapping

---

Qdrant

Import: from cocoindex.connectors import qdrant

Capabilities: Target only | Vector: Native | Model: Point-oriented

Connection Setup

from cocoindex.connectors import qdrant

from qdrant_client import QdrantClient
QDRANT_DB = coco.ContextKey[QdrantClient]("qdrant_db")

@coco.lifespan
def coco_lifespan(builder: coco.EnvironmentBuilder) -> Iterator[None]:
    client = qdrant.create_client("http://localhost:6333")
    builder.provide(QDRANT_DB, client)
    yield

As Target (Single Vector)

collection = await qdrant.mount_collection_target(
    QDRANT_DB,
    collection_name="embeddings",
    schema=await qdrant.CollectionSchema.create(
        vectors=qdrant.QdrantVectorDef(schema=EMBEDDER, distance="cosine"),
    ),
)

collection.declare_point(point=qdrant.PointStruct(
    id="point-1",
    vector=embedding_array.tolist(),
    payload={"text": "hello"},
))

As Target (Named Vectors)

schema = await qdrant.CollectionSchema.create(
    vectors={
        "text": qdrant.QdrantVectorDef(schema=text_embedder_key, distance="cosine"),
        "image": qdrant.QdrantVectorDef(schema=image_embedder_key, distance="cosine"),
    },
)
collection = await qdrant.mount_collection_target(QDRANT_DB, "multimodal", schema=schema)

collection.declare_point(point=qdrant.PointStruct(
    id="point-1",
    vector={"text": text_vec.tolist(), "image": image_vec.tolist()},
    payload={"title": "example"},
))

Distance Metrics

• "cosine" -- Cosine similarity (default)
• "dot" -- Dot product
• "euclid" -- Euclidean distance (L2)

---

SurrealDB

Import: from cocoindex.connectors import surrealdb

Capabilities: Target only | Vector: Native | Model: Graph + document

Connection Setup

from cocoindex.connectors import surrealdb

SURREAL_DB = coco.ContextKey[surrealdb.ConnectionFactory]("surreal_db")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    factory = surrealdb.ConnectionFactory(
        url="ws://localhost:8000/rpc",
        namespace="my_ns",
        database="my_db",
        credentials={"username": "root", "password": "root"},
    )
    builder.provide(SURREAL_DB, factory)
    yield

As Target (Table)

target_table = await surrealdb.mount_table_target(
    SURREAL_DB,
    table_name="records",
    table_schema=await surrealdb.TableSchema.from_class(MyRecord),
)

target_table.declare_row(row=MyRecord(id="rec1", name="example"))

As Target (Relation)

relation_target = await surrealdb.mount_relation_target(
    SURREAL_DB,
    table_name="likes",
    from_table=users_table,
    to_table=items_table,
    table_schema=await surrealdb.TableSchema.from_class(MyRelation),
)
relation_target.declare_relation(from_id="user:1", to_id="item:1", record=MyRelation(...))

---

LocalFS

Import: from cocoindex.connectors import localfs

Capabilities: Source + Target

As Source

from cocoindex.connectors import localfs
from cocoindex.resources.file import PatternFilePathMatcher

files = localfs.walk_dir(
    pathlib.Path("./data"),
    recursive=True,
    path_matcher=PatternFilePathMatcher(
        included_patterns=["**/*.py", "**/*.md"],
        excluded_patterns=[".*/**", "__pycache__/**"],
    ),
    live=True,  # Enable file watching for live mode
)

# Process each file
await coco.mount_each(process_file, files.items(), target_table)

As Target (Single File)

localfs.declare_file(
    outdir / "output.txt",
    "file content",
    create_parent_dirs=True,
)

As Target (Directory)

dir_target = await localfs.mount_dir_target(pathlib.Path("./output"))
dir_target.declare_file("file1.txt", "content 1")
dir_target.declare_file("subdir/file2.txt", "content 2")

Stable File Paths

Use localfs.FilePath for stable memoization when the base directory might move:

files = localfs.walk_dir(localfs.FilePath(path="./data"), ...)

---

Amazon S3

Import: from cocoindex.connectors import amazon_s3

Capabilities: Source only

Connection Setup

import aiobotocore.session
from aiobotocore.client import AioBaseClient

S3_CLIENT = coco.ContextKey[AioBaseClient]("s3_client")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    session = aiobotocore.session.get_session()
    async with session.create_client("s3") as s3_client:
        builder.provide(S3_CLIENT, s3_client)
        yield

Reading Objects

from cocoindex.connectors import amazon_s3

client = coco.use_context(S3_CLIENT)

# List and iterate objects
walker = amazon_s3.list_objects(
    client, "my-bucket",
    prefix="data/",
    path_matcher=PatternFilePathMatcher(included_patterns=["**/*.json"]),
    max_file_size=10_000_000,
)

await coco.mount_each(process_file, walker.items(), target)

# Single object
file = await amazon_s3.get_object(client, "s3://my-bucket/path/to/file.json")
content = await amazon_s3.read(client, "s3://my-bucket/path/to/file.json")

---

Kafka

Import: from cocoindex.connectors import kafka

Capabilities: Source + Target

As Source (Live)

from confluent_kafka.aio import AIOConsumer
from cocoindex.connectors import kafka

consumer = AIOConsumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "my-group",
    "enable.auto.commit": "false",
    "auto.offset.reset": "earliest",
})

items = kafka.topic_as_map(consumer, ["my-topic"])
await coco.mount_each(process_message, items, target_table)

topic_as_map() returns a LiveMapFeed, which mount_each() auto-detects for live mode.

As Target

from confluent_kafka.aio import AIOProducer

KAFKA_PRODUCER = coco.ContextKey[AIOProducer]("kafka_producer")

topic_target = await kafka.mount_kafka_topic_target(KAFKA_PRODUCER, "my-topic")
topic_target.declare_target_state(key="msg-key", value=json.dumps(data))

---

Apache Doris

Import: from cocoindex.connectors import doris

Capabilities: Target only | Vector: Native | Features: Vector indexes, inverted indexes, Stream Load

Connection Setup

from cocoindex.connectors import doris

DORIS_DB = coco.ContextKey[doris.ManagedConnection]("doris_db")

@coco.lifespan
def coco_lifespan(builder: coco.EnvironmentBuilder) -> Iterator[None]:
    config = doris.DorisConnectionConfig(
        fe_host="localhost",
        database="my_db",
        fe_http_port=8080,
        query_port=9030,
        username="root",
        password="",
    )
    conn = doris.connect(config)
    builder.provide(DORIS_DB, conn)
    yield

As Target

target_table = await doris.mount_table_target(
    DORIS_DB,
    table_name="embeddings",
    table_schema=await doris.TableSchema.from_class(MyRecord, primary_key=["id"]),
    vector_indexes=[doris.VectorIndexDef(field_name="embedding", metric_type="cosine_distance")],
)

target_table.declare_row(row=MyRecord(id=1, text="hello", embedding=vec))

---

Google Drive

Import: from cocoindex.connectors import google_drive

Capabilities: Source only

from cocoindex.connectors import google_drive

source = google_drive.GoogleDriveSource(
    service_account_credential_path="credentials.json",
    root_folder_ids=["folder-id"],
)
await coco.mount_each(process_file, source.items(), target)

---

Connector Comparison

---

See Also

• API Reference
• Patterns
• Setup Database

CocoIndex Common Patterns

Common patterns and workflows for building CocoIndex pipelines.

Core Pattern: TargetState = Transform(SourceState)

All CocoIndex pipelines follow this declarative pattern:

1. Read source state 2. Transform 3. Declare target state

CocoIndex handles the incremental sync automatically.

---

Pattern 1: File Transformation Pipeline

Use case: Transform files from one format to another (e.g., Markdown -> HTML, PDF -> Markdown)

import pathlib
import cocoindex as coco
from cocoindex.connectors import localfs
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from markdown_it import MarkdownIt

_markdown_it = MarkdownIt("gfm-like")

@coco.fn(memo=True)
async def process_file(file: FileLike, outdir: pathlib.Path) -> None:
    html = _markdown_it.render(await file.read_text())
    outname = "__".join(file.file_path.path.parts) + ".html"
    localfs.declare_file(outdir / outname, html, create_parent_dirs=True)

@coco.fn
async def app_main(sourcedir: pathlib.Path, outdir: pathlib.Path) -> None:
    files = localfs.walk_dir(
        sourcedir,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]),
        live=True,  # Enable live file watching
    )
    await coco.mount_each(process_file, files.items(), outdir)

app = coco.App(
    coco.AppConfig(name="FilesTransform"),
    app_main,
    sourcedir=pathlib.Path("./data"),
    outdir=pathlib.Path("./output_html"),
)

Key points:

• memo=True -- Skip reprocessing unchanged files
• mount_each() -- One component per file; keys from files.items() become component subpaths
• live=True on walk_dir() -- Enables file watching for live mode
• Auto-cleanup -- Deleting source file automatically removes output file

---

Pattern 2: Vector Embedding Pipeline

Use case: Chunk and embed documents for semantic search

import pathlib
from dataclasses import dataclass
from typing import AsyncIterator, Annotated

import asyncpg
from numpy.typing import NDArray

import cocoindex as coco
from cocoindex.connectors import localfs, postgres
from cocoindex.ops.text import RecursiveSplitter
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder
from cocoindex.resources.chunk import Chunk
from cocoindex.resources.file import FileLike, PatternFilePathMatcher
from cocoindex.resources.id import IdGenerator

DATABASE_URL = "postgres://cocoindex:cocoindex@localhost/cocoindex"
PG_DB = coco.ContextKey[asyncpg.Pool]("pg_db")
EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")

_splitter = RecursiveSplitter()

@dataclass
class DocEmbedding:
    id: int
    filename: str
    chunk_start: int
    chunk_end: int
    text: str
    embedding: Annotated[NDArray, EMBEDDER]

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with await asyncpg.create_pool(DATABASE_URL) as pool:
        builder.provide(PG_DB, pool)
        builder.provide(EMBEDDER, SentenceTransformerEmbedder("all-MiniLM-L6-v2"))
        yield

@coco.fn
async def process_chunk(
    chunk: Chunk, filename: pathlib.PurePath,
    id_gen: IdGenerator, table: postgres.TableTarget[DocEmbedding],
) -> None:
    table.declare_row(row=DocEmbedding(
        id=await id_gen.next_id(chunk.text),
        filename=str(filename),
        chunk_start=chunk.start.char_offset,
        chunk_end=chunk.end.char_offset,
        text=chunk.text,
        embedding=await coco.use_context(EMBEDDER).embed(chunk.text),
    ))

@coco.fn(memo=True)
async def process_file(file: FileLike, table: postgres.TableTarget[DocEmbedding]) -> None:
    text = await file.read_text()
    chunks = _splitter.split(text, chunk_size=2000, chunk_overlap=500, language="markdown")
    id_gen = IdGenerator()
    await coco.map(process_chunk, chunks, file.file_path.path, id_gen, table)

@coco.fn
async def app_main(sourcedir: pathlib.Path) -> None:
    target_table = await postgres.mount_table_target(
        PG_DB,
        table_name="doc_embeddings",
        table_schema=await postgres.TableSchema.from_class(DocEmbedding, primary_key=["id"]),
    )
    target_table.declare_vector_index(column="embedding")

    files = localfs.walk_dir(sourcedir, recursive=True,
        path_matcher=PatternFilePathMatcher(included_patterns=["**/*.md"]))
    await coco.mount_each(process_file, files.items(), target_table)

app = coco.App(coco.AppConfig(name="TextEmbedding"), app_main,
    sourcedir=pathlib.Path("./markdown_files"))

Key points:

• mount_table_target(PG_DB, ...) -- Takes ContextKey as first argument
• Annotated[NDArray, EMBEDDER] -- Vector annotation uses ContextKey for auto dimension inference
• map() -- Concurrent execution within a component (no child components created)
• IdGenerator -- Generates stable unique IDs for chunks across incremental updates
• memo=True on process_file -- Skips unchanged files entirely

---

Pattern 3: Database Source -> Transform -> Database Target

Use case: Transform data from one database to another

from dataclasses import dataclass
from typing import AsyncIterator

import asyncpg
import cocoindex as coco
from cocoindex.connectors import postgres

SOURCE_DB_URL = "postgres://localhost/source_db"
TARGET_DB_URL = "postgres://localhost/target_db"

SOURCE_DB = coco.ContextKey[asyncpg.Pool]("source_db")
TARGET_DB = coco.ContextKey[asyncpg.Pool]("target_db")

@dataclass
class SourceRecord:
    id: int
    name: str
    value: float

@dataclass
class TargetRecord:
    id: int
    name: str
    value: float
    processed: bool

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    async with (
        await asyncpg.create_pool(SOURCE_DB_URL) as source_pool,
        await asyncpg.create_pool(TARGET_DB_URL) as target_pool,
    ):
        builder.provide(SOURCE_DB, source_pool)
        builder.provide(TARGET_DB, target_pool)
        yield

@coco.fn(memo=True)
async def process_record(record: SourceRecord, target_table: postgres.TableTarget[TargetRecord]) -> None:
    target_table.declare_row(row=TargetRecord(
        id=record.id, name=record.name.upper(),
        value=record.value * 2, processed=True,
    ))

@coco.fn
async def app_main() -> None:
    target_table = await postgres.mount_table_target(
        TARGET_DB,
        table_name="target_records",
        table_schema=await postgres.TableSchema.from_class(TargetRecord, primary_key=["id"]),
    )

    source = postgres.PgTableSource(
        coco.use_context(SOURCE_DB),
        table_name="source_records",
        row_type=SourceRecord,
    )
    await coco.mount_each(
        coco.component_subpath("record"),
        process_record,
        source.fetch_rows().items(key=lambda r: r.id),
        target_table,
    )

app = coco.App(coco.AppConfig(name="DatabaseTransform"), app_main)

---

Pattern 4: LLM-Based Extraction Pipeline

Use case: Extract structured data using LLMs

import instructor
from dataclasses import dataclass
from typing import AsyncIterator
from pydantic import BaseModel
from litellm import acompletion

import asyncpg
import cocoindex as coco
from cocoindex.connectors import postgres

DATABASE_URL = "postgres://cocoindex:cocoindex@localhost/cocoindex"
PG_DB = coco.ContextKey[asyncpg.Pool]("pg_db")

_instructor_client = instructor.from_litellm(acompletion, mode=instructor.Mode.JSON)

class ExtractedTopic(BaseModel):
    name: str
    description: str

class ExtractionResult(BaseModel):
    title: str
    topics: list[ExtractedTopic]

@dataclass
class Message:
    id: int
    title: str
    content: str

@dataclass
class Topic:
    message_id: int
    name: str
    description: str

@coco.fn(memo=True)
async def extract_and_store(
    content: str, message_id: int,
    messages_table: postgres.TableTarget[Message],
    topics_table: postgres.TableTarget[Topic],
) -> None:
    result = await _instructor_client.chat.completions.create(
        model="gpt-4",
        response_model=ExtractionResult,
        messages=[{"role": "user", "content": f"Extract topics:\n\n{content}"}],
    )
    messages_table.declare_row(row=Message(id=message_id, title=result.title, content=content))
    for topic in result.topics:
        topics_table.declare_row(row=Topic(
            message_id=message_id, name=topic.name, description=topic.description,
        ))

@coco.fn
async def app_main(input_texts: list[str]) -> None:
    messages_table = await postgres.mount_table_target(
        PG_DB, table_name="messages",
        table_schema=await postgres.TableSchema.from_class(Message, primary_key=["id"]),
    )
    topics_table = await postgres.mount_table_target(
        PG_DB, table_name="topics",
        table_schema=await postgres.TableSchema.from_class(Topic, primary_key=["message_id", "name"]),
    )
    for idx, text in enumerate(input_texts):
        await coco.mount(
            coco.component_subpath("text", idx),
            extract_and_store, text, idx, messages_table, topics_table,
        )

app = coco.App(coco.AppConfig(name="LLMExtraction"), app_main, input_texts=["text1...", "text2..."])

Key points:

• memo=True on extraction -- Avoids re-calling LLM for unchanged inputs
• Multiple target tables -- Declare rows in multiple tables from single component
• Pydantic models -- For structured LLM outputs

---

Pattern 5: Kafka Streaming Pipeline

Use case: Process Kafka messages and write to a database

import json
from collections.abc import AsyncIterator
from dataclasses import dataclass

from confluent_kafka import Message
from confluent_kafka.aio import AIOConsumer

import cocoindex as coco
from cocoindex.connectors import kafka, lancedb

LANCE_DB = coco.ContextKey[lancedb.LanceAsyncConnection]("lance_db")

@dataclass
class Product:
    sku: str
    name: str
    category: str
    price: float

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    conn = await lancedb.connect_async("./lancedb_data")
    builder.provide(LANCE_DB, conn)
    yield

@coco.fn
async def process_message(msg: Message, table: lancedb.TableTarget[Product]) -> None:
    value = msg.value()
    if value is None:
        return
    row = json.loads(value.decode() if isinstance(value, bytes) else value)
    table.declare_row(row=Product(**{**row, "price": float(row["price"])}))

@coco.fn
async def app_main() -> None:
    products_table = await lancedb.mount_table_target(
        LANCE_DB, table_name="products",
        table_schema=await lancedb.TableSchema.from_class(Product, primary_key=["sku"]),
    )

    consumer = AIOConsumer({
        "bootstrap.servers": "localhost:9092",
        "group.id": "my-group",
        "enable.auto.commit": "false",
        "auto.offset.reset": "earliest",
    })
    items = kafka.topic_as_map(consumer, ["products-topic"])
    await coco.mount_each(process_message, items, products_table)

app = coco.App(coco.AppConfig(name="KafkaToLanceDB"), app_main)

Key points:

• kafka.topic_as_map() returns a LiveMapFeed -- mount_each() auto-detects for live mode
• Pipeline runs continuously in live mode, processing new messages as they arrive

---

Pattern 6: Context Management for Shared Resources

Use case: Share expensive resources across components

import cocoindex as coco
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

EMBEDDER = coco.ContextKey[SentenceTransformerEmbedder]("embedder")
CONFIG = coco.ContextKey[dict]("config")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    builder.provide(EMBEDDER, SentenceTransformerEmbedder("all-MiniLM-L6-v2"))
    builder.provide(CONFIG, {"chunk_size": 1000, "overlap": 200})
    yield

@coco.fn
async def process_item(text: str) -> None:
    embedder = coco.use_context(EMBEDDER)
    config = coco.use_context(CONFIG)
    embedding = await embedder.embed(text)
    ...

Key points:

• detect_change=True -- Opt in to invalidate memos when value changes (use for models/config affecting output)
• detect_change=False (default) -- For resources not affecting computation (DB connections, loggers)
• ContextKey name is stable identity -- avoid renaming across runs

---

Common Anti-Patterns to Avoid

Missing @coco.fn Decorator

# BAD: Missing decorator
async def process_file(file, table):
    table.declare_row(...)

# GOOD:
@coco.fn
async def process_file(file, table):
    table.declare_row(...)

Reprocessing Everything

# BAD: No memoization
@coco.fn  # Missing memo=True
async def process_file(file, table):
    embedding = await embedder.embed(await file.read_text())  # Expensive!

# GOOD:
@coco.fn(memo=True)
async def process_file(file, table):
    ...

Unstable Component Paths

# BAD: Using object references or indices
for file in files:
    await coco.mount(coco.component_subpath(file), ...)     # Object ref
for idx, item in enumerate(items):
    await coco.mount(coco.component_subpath(idx), ...)      # Index changes

# GOOD: Use stable identifiers
await coco.mount_each(process_file, files.items(), table)   # Keys from items()

Loading Resources Per Component

# BAD: Loading model in every component
@coco.fn
async def process(text):
    model = SentenceTransformer("model")  # Loaded repeatedly!

# GOOD: Load once in lifespan, use via context
embedder = coco.use_context(EMBEDDER)

Mixing Target State with Side Effects

# BAD: Side effects not detected
@coco.fn
async def process(data):
    requests.post("https://api.example.com", json=data)  # Not detected!

# GOOD: Only declare target states
table.declare_row(row=result)

---

See Also

• Connectors Reference
• API Reference
• Setup Project

Database Setup Guide

Setup instructions for databases used with CocoIndex.

PostgreSQL with pgvector

Using Docker (Recommended)

cat > docker-compose.yml <<EOF
version: '3.8'

services:
  postgres:
    image: pgvector/pgvector:pg16
    container_name: cocoindex-postgres
    environment:
      POSTGRES_USER: cocoindex
      POSTGRES_PASSWORD: cocoindex
      POSTGRES_DB: cocoindex
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U cocoindex"]
      interval: 5s
      timeout: 5s
      retries: 5

volumes:
  postgres_data:
EOF

docker-compose up -d

Connection URL: postgres://cocoindex:cocoindex@localhost:5432/cocoindex

Using Existing PostgreSQL

psql "postgres://user:password@localhost/dbname" \
  -c "CREATE EXTENSION IF NOT EXISTS vector;"

Environment Configuration

# .env
POSTGRES_URL=postgres://cocoindex:cocoindex@localhost:5432/cocoindex

---

SQLite with sqlite-vec

Installation

pip install sqlite-vec

Note for macOS: Use Homebrew Python for extension support:

brew install python

Usage in Code

from cocoindex.connectors import sqlite

conn = sqlite.connect("./data.db", load_vec="auto")  # Auto-load sqlite-vec
conn = sqlite.connect("./data.db", load_vec=True)    # Require sqlite-vec
conn = sqlite.connect("./data.db", load_vec=False)   # No vector support

---

LanceDB

Installation

pip install lancedb

Local Storage

No setup needed. Just specify a directory:

conn = await lancedb.connect_async("./lancedb_data")

Cloud Storage (S3, GCS)

# AWS S3
conn = await lancedb.connect_async("s3://your-bucket/lancedb_data")

# Google Cloud Storage
conn = await lancedb.connect_async("gs://your-bucket/lancedb_data")

Configure credentials via environment variables (AWS_ACCESS_KEY_ID, GOOGLE_APPLICATION_CREDENTIALS, etc.).

---

Qdrant

Using Docker

cat > docker-compose.yml <<EOF
version: '3.8'

services:
  qdrant:
    image: qdrant/qdrant:latest
    container_name: cocoindex-qdrant
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage

volumes:
  qdrant_data:
EOF

docker-compose up -d

Using Qdrant Cloud

client = qdrant.create_client(
    url="https://your-cluster.qdrant.io",
    api_key="your-api-key",
    prefer_grpc=True,
)

Environment Configuration

# .env
QDRANT_URL=http://localhost:6333
# QDRANT_API_KEY=your-key  # For Qdrant Cloud

---

SurrealDB

Using Docker

docker run --rm -p 8000:8000 surrealdb/surrealdb:latest \
  start --user root --pass root

Usage in Code

factory = surrealdb.ConnectionFactory(
    url="ws://localhost:8000/rpc",
    namespace="my_ns",
    database="my_db",
    credentials={"username": "root", "password": "root"},
)

---

Multi-Database Setup

# docker-compose.yml
version: '3.8'

services:
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: cocoindex
      POSTGRES_PASSWORD: cocoindex
      POSTGRES_DB: cocoindex
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage

volumes:
  postgres_data:
  qdrant_data:

# .env
POSTGRES_URL=postgres://cocoindex:cocoindex@localhost:5432/cocoindex
QDRANT_URL=http://localhost:6333

---

Common Issues

PostgreSQL Connection Refused

docker-compose ps          # Check if running
nc -zv localhost 5432      # Check port

pgvector Extension Not Found

psql -U postgres -c "SELECT * FROM pg_available_extensions WHERE name = 'vector';"

Use the pgvector/pgvector:pg16 Docker image which includes it.

SQLite Extensions Not Loading

import sqlite3
conn = sqlite3.connect(":memory:")
conn.enable_load_extension(True)  # Should not raise

If it fails, use Homebrew Python on macOS or ensure Python was built with --enable-loadable-sqlite-extensions on Linux.

Qdrant Connection Issues

curl http://localhost:6333/       # Check health
docker logs cocoindex-qdrant      # Check logs

---

See Also

• Connectors Reference
• Patterns Reference

Project Setup Guide

Setting up CocoIndex projects for different use cases.

Creating a New Project

cocoindex init my-project
cd my-project

This creates: main.py, pyproject.toml, .env, README.md.

pip install -e .

Dependencies by Use Case

Vector Embedding Pipeline

[project]
dependencies = [
    "cocoindex>=1.0.0",
    "sentence-transformers",
    "asyncpg",
]

PostgreSQL Integration

[project]
dependencies = [
    "cocoindex>=1.0.0",
    "asyncpg",
]

SQLite Integration

[project]
dependencies = [
    "cocoindex>=1.0.0",
    "sqlite-vec",
]

LanceDB Integration

[project]
dependencies = [
    "cocoindex>=1.0.0",
    "lancedb",
]

Qdrant Integration

[project]
dependencies = [
    "cocoindex>=1.0.0",
    "qdrant-client",
]

Kafka Integration

[project]
dependencies = [
    "cocoindex>=1.0.0",
    "confluent-kafka",
]

LLM-Based Extraction

[project]
dependencies = [
    "cocoindex>=1.0.0",
    "litellm",
    "instructor",
    "pydantic>=2.0",
    "asyncpg",
]

---

Environment Configuration

.env File

CocoIndex automatically loads .env from the current directory.

# CocoIndex internal database (required)
COCOINDEX_DB=./cocoindex.db

# PostgreSQL (if using)
POSTGRES_URL=postgres://user:pass@localhost/db

# Qdrant (if using)
QDRANT_URL=http://localhost:6333

# API keys (if using LLM extraction)
OPENAI_API_KEY=your-openai-api-key
ANTHROPIC_API_KEY=your-anthropic-api-key

Manual Settings (in lifespan)

@coco.lifespan
def coco_lifespan(builder: coco.EnvironmentBuilder) -> Iterator[None]:
    builder.settings.db_path = pathlib.Path("./custom.db")
    yield

---

Running Your Pipeline

pip install -e .                    # Install dependencies
cocoindex update main.py            # Run pipeline
cocoindex update main.py -L         # Run in live mode
cocoindex show main.py              # Show component paths
cocoindex drop main.py -f           # Reset everything

---

Common Issues

Import Errors

pip install -e .

Database Connection Errors

Verify database is running and .env has correct URLs. See setup_database.md.

---

See Also

• Database Setup
• Patterns
• API Reference