Cocoindex Use Cases in Software Development: Declarative Data Pipelines for ETL, ML, and Live Streaming

Cocoindex is a declarative, change-aware execution engine that enables software developers to build incremental data pipelines for ETL, document processing, ML feature extraction, and live streaming without manually managing state or change detection logic.

The cocoindex-io/cocoindex repository provides a Python framework backed by a Rust core that treats data pipelines as declarative descriptions of target states rather than imperative scripts. By combining memoization, stable component paths, and atomic state diffs, Cocoindex eliminates boilerplate code for tracking changes and synchronizing external resources, making it particularly effective for modern software development workflows.

Core Architecture: App, Component, and Target State

Cocoindex organizes code into three interlocking abstractions that enable declarative execution:

• App – The top-level runnable unit defined in cocoindex._internal.app.App that owns a root component and orchestrates pipeline updates.
• Component – A processing unit created via mount() or use_mount() in cocoindex._internal.api.mount, each owning a stable component path that persists across runs.
• Target State – A declarative description of external resources (files, database rows, vector store entries) defined in cocoindex._internal.target_state.TargetState, which the engine diffs against previous runs to apply only necessary creates, updates, or deletes.

When a pipeline runs, the engine computes a stable path (StablePath) for each component and fingerprints the logic using AST-based analysis. This allows Cocoindex to determine exactly which components changed and require re-execution.

Declarative → Incremental → Atomic Execution

Cocoindex pipelines follow a three-phase execution model implemented across the Python layer and the Rust core:

1. Declarative – Developers use high-level APIs like coco.mount, coco.use_mount, and coco.mount_target to declare what the final data state should look like, eliminating explicit "write" or "update" logic.
2. Incremental – The engine records fingerprints of function logic and input data via cocoindex._internal.function.SyncFunction.__call__ and _compute_logic_fingerprint. Only components with changed inputs or logic are re-executed.
3. Atomic – The Rust core (rust/core) applies each component's target-state diff in a single transaction. If a component disappears, its target states are automatically cleaned up.

This architecture ensures that pipelines are deterministic, efficient, and safe to rerun without side effects.

Top Cocoindex Use Cases in Software Development

ETL and Document Processing Pipelines

Cocoindex excels at extracting, transforming, and loading file-based data. The framework's localfs connector, combined with memoized functions, creates efficient document processing workflows.

The files_transform example demonstrates converting Markdown files to HTML incrementally:

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

_md = MarkdownIt("gfm-like")

@coco.fn(memo=True)
async def process_file(file: FileLike, outdir: pathlib.Path) -> None:
    html = _md.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,
    )
    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 implementation details:

• @coco.fn(memo=True) deduplicates work on unchanged files by storing fingerprints in cocoindex._internal.memo_fingerprint.py.
• mount_each creates a component per Markdown file, each with its own stable path.
• The localfs connector declares output files as target states; only new or changed files are written.

Source: [examples/files_transform/main.py](https://github.com/cocoindex-io/cocoindex/blob/main/examples/files_transform/main.py)

ML Feature Extraction and Vector Search

For machine learning workflows, Cocoindex provides automatic memoization of expensive model calls and native connectors for vector databases like Qdrant.

import pathlib
import cocoindex as coco
from cocoindex.connectors import qdrant
from cocoindex.ops.sentence_transformers import SentenceTransformerEmbedder

embedder = SentenceTransformerEmbedder(model_name="all-MiniLM-L6-v2")

@coco.fn(memo=True)
async def embed_file(file: FileLike, target: qdrant.CollectionTarget) -> None:
    txt = await file.read_text()
    vec = embedder.embed(txt)
    target.declare_vector(
        id=file.file_path.path.as_posix(),
        vector=vec,
        payload={"source": str(file.file_path)},
    )

@coco.fn
async def app_main(sourcedir: pathlib.Path, collection: str) -> None:
    files = localfs.walk_dir(sourcedir, live=True)
    target = await coco.use_mount(qdrant.declare_collection_target, collection)
    await coco.mount_each(embed_file, files.items(), target)

coco.App("DocsEmbedding", app_main,
          sourcedir=pathlib.Path("./docs"),
          collection="my_docs").update_blocking(report_to_stdout=True)

The SentenceTransformerEmbedder runs only when inputs change, and embeddings are stored declaratively via declare_vector. The Qdrant connector handles the complexity of upserting vectors only when the source document content changes.

Source: [examples/text_embedding_qdrant/main.py](https://github.com/cocoindex-io/cocoindex/blob/main/examples/text_embedding_qdrant/main.py)

Live Streaming and Event Processing

Cocoindex supports continuous data sources through LiveComponent, which runs in the background and processes incremental updates without restarting the pipeline.

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

class KafkaToLance(coco.LiveComponent):
    def __init__(self, topic: str, collection: str):
        self.topic = topic
        self.collection = collection

    async def process_live(self, operator: coco.LiveComponentOperator) -> None:
        async for msg in kafka.consume(self.topic):
            summary = await litellm.summarize(msg.value)
            await operator.target.declare_row(id=msg.offset, payload=summary)

async def app_main(topic: str, collection: str) -> None:
    target = await coco.use_mount(lancedb.declare_collection_target, collection)
    await coco.mount(KafkaToLance(topic, collection), target)

coco.App("KafkaLance", app_main, topic="events", collection="events").update()

Live components are instantiated via coco.mount when is_live_component_class triggers the _mount_live_component path in cocoindex._internal.api.py. This enables use cases like Kafka-to-Database streaming, real-time log processing, and continuous file system monitoring.

Source: [examples/kafka_to_lancedb/main.py](https://github.com/cocoindex-io/cocoindex/blob/main/examples/kafka_to_lancedb/main.py)

Context-Based Dependency Injection

Cocoindex replaces global singletons with typed context keys (ContextKey) provided during the environment lifespan. The current component's context is stored in a ContextVar, making dependencies automatically available downstream.

Define a key:

PG_DB = coco.ContextKey[postgres.PgDatabase]("pg_db")

Provide it in the lifespan:

with coco.environment.builder() as env:
    env.provide(PG_DB, pg_pool)

Consume it inside a component:

db = coco.use_context(PG_DB)  # resolved from ComponentContext

Implementation details reside in cocoindex._internal.component_ctx and cocoindex._internal.api.

Extensible Connectors and Performance Optimization

Cocoindex ships with connectors for PostgreSQL, Qdrant, LanceDB, Google Drive, and local filesystems. Each connector implements a declare_*_target function returning a TargetStateProvider, making it straightforward to add new storage backends.

For performance:

• Runner abstraction – Offload CPU-heavy functions to thread or GPU pools using coco.runner.GPU or coco.runner.Runner.
• Automatic batching – Decorate async functions with @coco.fn(batching=True) to process multiple inputs together. The batching logic lives in cocoindex._internal.function.AsyncFunction.

Summary

• Cocoindex provides a declarative execution model where developers describe target states rather than writing imperative update logic.
• The engine automatically handles incremental updates through stable component paths, memoization, and logic fingerprinting in cocoindex._internal.function and cocoindex._internal.stable_path.
• LiveComponent enables continuous processing of streaming data sources like Kafka without pipeline restarts.
• Context-based dependency injection replaces global state with typed keys provided via coco.environment.builder().
• The Rust core guarantees atomic application of state diffs and automatic cleanup of removed components.
• Built-in connectors for vector stores, databases, and filesystems make Cocoindex suitable for ETL, ML feature extraction, and real-time data synchronization.

Frequently Asked Questions

What is Cocoindex used for in software development?

Cocoindex is used to build declarative data pipelines that synchronize data between sources and destinations. Common use cases include ETL workflows (transforming Markdown to HTML), ML feature extraction (generating and caching text embeddings), incremental builds (processing only changed source files), and live streaming (moving data from Kafka to vector databases). The framework handles change detection, memoization, and atomic updates automatically.

How does Cocoindex handle incremental updates?

Cocoindex computes stable paths and fingerprints for every component. When you decorate a function with @coco.fn(memo=True), the engine stores a hash of the function's AST logic and input arguments in cocoindex._internal.memo_fingerprint.py. On subsequent runs, only components with changed fingerprints are re-executed, making pipelines efficient even with large datasets.

What is the difference between mount() and mount_each()?

mount() creates a single component from a class or function, suitable for singleton resources or live components. mount_each() creates multiple components from an iterable, generating a unique component with its own stable path for each item. Use mount_each() when processing collections like directories of files or database result sets, as shown in the text embedding and file transform examples.

Can Cocoindex process live streaming data?

Yes. Cocoindex provides LiveComponent, a subclass that runs continuously in the background. When you subclass coco.LiveComponent and implement process_live(), the runtime keeps the component alive after the initial mark_ready call. This enables processing of streaming sources like Kafka topics, file system watches, or message queues without tearing down and rebuilding the entire pipeline.

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