Neo4j connector

Write to Neo4j (a property graph database) with support for node tables, relationship tables (edges), per-database multitenancy, atomic-batch writes via Bolt transactions, real CREATE CONSTRAINT uniqueness, and vector indexes with cosine / euclidean similarity.

Version

v 1.0.14

Last reviewed

Nov 4, 2026

The neo4j connector writes records to Neo4j, a property graph database. It supports node tables (labels), relationship tables (edge types), per-database multitenancy (one Neo4j cluster, many isolated databases), real Cypher uniqueness constraints, and vector indexes via the CREATE VECTOR INDEX DDL form.

from cocoindex.connectors import neo4j

pip install cocoindex[neo4j]

Connection setup

Create a ConnectionFactory and provide it via a ContextKey. The factory holds the Bolt URI, optional auth, and the target database name; it opens one connection-pooled Neo4j async driver per event loop, on first use on that loop, shared by all tables and batches. Call await factory.close() at shutdown (in the lifespan, after yield) to close it.

from collections.abc import AsyncIterator
from cocoindex.connectors import neo4j
import cocoindex as coco

KG_DB: coco.ContextKey[neo4j.ConnectionFactory] = coco.ContextKey("kg_db")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    factory = neo4j.ConnectionFactory(
        uri="bolt://localhost:7687",
        auth=("neo4j", "cocoindex"),
        database="neo4j",
    )
    builder.provide(KG_DB, factory)
    yield
    await factory.close()

auth is optional — omit it for unauthenticated dev instances. database defaults to "neo4j" (the default db that ships with every Neo4j 5 installation).

Multitenancy

A single Neo4j cluster can host many isolated databases. Pair each database with its own ContextKey and ConnectionFactory(database=...):

KG_DB: coco.ContextKey[neo4j.ConnectionFactory] = coco.ContextKey("kg_db")
APIS_DB: coco.ContextKey[neo4j.ConnectionFactory] = coco.ContextKey("apis_db")

@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    uri = "bolt://localhost:7687"
    auth = ("neo4j", "cocoindex")
    kg_factory = neo4j.ConnectionFactory(uri=uri, auth=auth, database="kg")
    apis_factory = neo4j.ConnectionFactory(uri=uri, auth=auth, database="apis")
    builder.provide(KG_DB, kg_factory)
    builder.provide(APIS_DB, apis_factory)
    yield
    await kg_factory.close()
    await apis_factory.close()

Different ContextKeys with different database names produce fully separate target-state trees — changes to one never spill into the other.

As target

The neo4j connector provides target state APIs for writing nodes to node tables and edges to relationship tables. CocoIndex tracks what records should exist and automatically handles upserts and deletions.

Each apply batch is wrapped in a single Neo4j transaction (tx.commit() on success, rollback on exception), so partial writes never leak into the database. Within a batch, writes are ordered as node upserts → relation upserts → relation deletes → node deletes so dependent edges always see their endpoints. A batch that fails with a transient error (typically a deadlock between concurrent batches merging edges onto the same nodes) is rolled back and retried with backoff before the failure is reported.

Declaring target states

Node tables (parent state)

Declares a node label as a target state. Returns a NodeTarget for declaring nodes.

def declare_node_target(
    db: ContextKey,
    table_name: str,
    table_schema: TableSchema[RowT] | None = None,
    *,
    primary_key: str = "id",
    managed_by: Literal["system", "user"] = "system",
) -> NodeTarget[RowT, coco.PendingS]

Parameters:

• db — A ContextKey[neo4j.ConnectionFactory] for the Neo4j connection.
• table_name — The Cypher node label (e.g. "Document"). It must not contain __ or end with _: constraint and index names join the label to a property name with __, so such a label could share a name with another table’s.
• table_schema — Optional schema definition (see Table Schema). The schema participates in CocoIndex’s fingerprint (so two flows declaring the same label must agree); per-property type DDL is not emitted in v1.
• primary_key — Single property name used as the node’s primary key. Defaults to "id". Must match the schema’s primary key when table_schema is given. Compound primary keys are not supported in v1.0.
• managed_by — Whether CocoIndex manages the table lifecycle ("system") or assumes it exists ("user").

Returns: A pending NodeTarget. Use await neo4j.mount_node_target(KG_DB, ...) to get a resolved target.

Nodes (child states)

Once a NodeTarget is resolved, declare nodes to be upserted (translated to MERGE (n:Label {pk: $key_0}) SET n += $props):

def NodeTarget.declare_record(
    self,
    *,
    row: RowT,
) -> None

Parameters:

• row — A row object (dict, dataclass, NamedTuple, or Pydantic model). Must include the primary_key field declared above.

declare_row is an alias for declare_record, for compatibility with Postgres and other RDBMS targets.

Relationship tables (parent state)

Declares a relationship type as a target state. Returns a RelationshipTarget for declaring edges.

def declare_relationship_target(
    db: ContextKey,
    table_name: str,
    from_table: NodeTarget,
    to_table: NodeTarget,
    table_schema: TableSchema[RowT] | None = None,
    *,
    primary_key: str = "id",
    managed_by: Literal["system", "user"] = "system",
) -> RelationshipTarget[RowT, coco.PendingS]

Parameters:

• db — A ContextKey[neo4j.ConnectionFactory] for the Neo4j connection.
• table_name — The Cypher relationship type (e.g. "MENTION"). Same naming rule as a node label: no __, no trailing _. A node label and a relationship type may share a name.
• from_table — The node target whose nodes are the source endpoints of edges in this relationship. A TableTarget of the deprecated table API works too.
• to_table — The node target whose nodes are the target endpoints of edges in this relationship.
• table_schema — Optional schema for the relationship’s own properties. The relationship’s primary_key field uniquely identifies each edge.
• primary_key — Single property name used as the edge’s primary key. Defaults to "id". Must match the schema’s primary key when table_schema is given.
• managed_by — Whether CocoIndex manages the relationship lifecycle ("system") or assumes it exists ("user").

Returns: A pending RelationshipTarget. Use await neo4j.mount_relationship_target(KG_DB, ...) to get a resolved target.

Relationships (child states)

Once a RelationshipTarget is resolved, declare edges. Each declaration MERGEs the source endpoint, the target endpoint, then the relationship.

def RelationshipTarget.declare_relationship(
    self,
    *,
    from_id: Any,
    to_id: Any,
    record: RowT | None = None,
) -> None

Parameters:

• from_id — The source node’s primary-key value. The connector MERGEs (s:FromLabel {pk: $from_id}) so endpoints are auto-created if absent.
• to_id — The target node’s primary-key value. Same MERGE behavior.
• record — Optional row object whose fields populate the relationship’s properties. Its primary_key field, if present, is the edge’s id.

If record is omitted or has no primary key value, the connector derives the edge id from its endpoints: the source label, source id, target label and target id, each written as <UTF-8 byte length>:<value>;. For example, an edge from Person "alice" to Task 42 gets the id 6:Person;5:alice;4:Task;2:42;. A string id that reads as an integer, or that starts with s:, is written with an s: prefix, so the ids 1 and "1" never share an edge. Endpoint ids must then be int or str; for other types, pass a record that carries the primary key.

An edge id names one edge. When the endpoints of an edge with an explicit id change, the edge moves: the upsert deletes the relationship of this type with the same id between the old endpoints. To find it, every upsert looks the edge up by its primary key property. A system-managed relationship table has an index on it; for a managed_by="user" table, create one yourself.

Vector indexes (attachment)

Declares a vector index on a column of a node table. Vector indexes are an attachment to a NodeTarget:

def NodeTarget.declare_vector_index(
    self,
    *,
    name: str | None = None,
    field: str,
    metric: Literal["cosine", "euclidean"] = "cosine",
    dimension: int,
) -> None

Parameters:

• name — Optional logical name for the index. Defaults to f"vec_{table_name}__{field}".
• field — The node property holding the vector.
• metric — Similarity metric: "cosine" or "euclidean". Translated to Neo4j’s vector.similarity_function option.
• dimension — The vector’s dimension. Required.

The connector emits:

CREATE VECTOR INDEX `coco_vec_<Label>__<field>` IF NOT EXISTS
FOR (n:`Label`) ON n.`field`
OPTIONS { indexConfig: {
  `vector.dimensions`: <N>,
  `vector.similarity_function`: '<metric>'
} }

The Neo4j index is named after the label and field, not after name, so a field has at most one vector index. Renaming name keeps the index. Declaring a second name for the same field is an error.

Vectors are float32 only.

Table schema: from Python class

Build a TableSchema by introspecting a record type:

@classmethod
async def TableSchema.from_class(
    cls,
    record_type: type[RowT],
    *,
    primary_key: str = "id",
    column_overrides: dict[str, Neo4jType | VectorSchemaProvider] | None = None,
) -> TableSchema[RowT]

Parameters:

• record_type — A dataclass, NamedTuple, or Pydantic model.
• primary_key — Field name to use as the table’s primary key. Defaults to "id".
• column_overrides — Optional dict mapping field names to Neo4jType or VectorSchemaProvider to override the default Python-to-Neo4j type mapping.

Returns: A TableSchema[RowT] populated from the class’s fields.

Default Python → Neo4j type mapping

Most types pass through native Bolt encoding — no per-value transform applied:

Neo4jType

Override the default mapping for a single column with Neo4jType:

class Neo4jType(NamedTuple):
    neo4j_type: str
    encoder: ValueEncoder | None = None

Use with typing.Annotated:

from typing import Annotated
from dataclasses import dataclass
from cocoindex.connectors.neo4j import Neo4jType

@dataclass
class Row:
    id: str
    score: Annotated[float, Neo4jType("STRING", encoder=str)]

The neo4j_type string is metadata-only — it participates in the schema fingerprint (so two flows declaring the same table must agree) but no per-property type DDL is emitted from it.

VectorSchemaProvider

For NumPy ndarray columns, attach a VectorSchema annotation to specify dtype + dimension. See VectorSchema for details.

Table schema: explicit column definitions

Build a TableSchema directly from a dict of column definitions when the row type is dynamic:

from cocoindex.connectors.neo4j import TableSchema, ColumnDef

schema = TableSchema(
    columns={
        "filename": ColumnDef(type="STRING"),
        "title": ColumnDef(type="STRING"),
        "summary": ColumnDef(type="STRING", nullable=True),
    },
    primary_key="filename",
)

ColumnDef fields:

• type — The Neo4j type string (metadata only; see table above).
• nullable — Whether the column may be None. Defaults to True.
• encoder — Optional Callable[[Any], Any] applied to non-None values before they’re sent to Neo4j.

DDL: indexes and constraints

For each managed table, the connector creates supporting Cypher artifacts on first run:

• For node tables: a uniqueness constraint on the primary key —

  Copy

  CREATE CONSTRAINT `coco_uniq_<Label>__<pk>` IF NOT EXISTS
  FOR (n:`<Label>`) REQUIRE n.`<pk>` IS UNIQUE

  Neo4j auto-creates a backing index for each constraint, so a separate CREATE INDEX is redundant on nodes.
• For relationship tables:

  Copy

  CREATE INDEX `coco_idx_rel_<RelType>__<pk>` IF NOT EXISTS
  FOR ()-[r:`<RelType>`]-() ON (r.`<pk>`)

A system-managed table owns its label or relationship type, the way a Postgres table owns its rows. When the table is no longer declared, or on cocoindex drop, the connector destroys it: every node carrying the label is deleted (DETACH DELETE, so the relationships attached to those nodes go with them) or every relationship of the type is deleted, and then the constraint or index is dropped. A node table’s vector indexes (the coco_vec_<Label>__<field> indexes on its label) are dropped with it. This includes nodes and relationships written under that label or type by anything other than CocoIndex — to share a label with data CocoIndex doesn’t own, declare the table with managed_by="user".

The same teardown happens when a table’s identity changes — its primary key field, or a relationship table’s endpoint tables or their primary keys: the table is destroyed and recreated, and the components that declared its records are reprocessed. Its vector indexes are dropped and created again. Attaching a schema or changing a property type does not rebuild the table; at most the records are re-upserted.

When managed_by="user" is set, the connector skips DDL entirely (you’re responsible for creating and dropping the schema) and never bulk-deletes: when the table is no longer declared or the app is dropped, its nodes or relationships are left as-is (see Destroy vs. Abandon). Record-level upserts and deletes still work while the table is declared.

Example: Node tables

from collections.abc import AsyncIterator
from dataclasses import dataclass
import cocoindex as coco
from cocoindex.connectors import neo4j

KG_DB: coco.ContextKey[neo4j.ConnectionFactory] = coco.ContextKey("kg_db")


@dataclass
class Document:
    filename: str
    title: str
    summary: str


@coco.lifespan
async def coco_lifespan(builder: coco.EnvironmentBuilder) -> AsyncIterator[None]:
    factory = neo4j.ConnectionFactory(
        uri="bolt://localhost:7687",
        auth=("neo4j", "cocoindex"),
        database="neo4j",
    )
    builder.provide(KG_DB, factory)
    yield
    await factory.close()


@coco.fn
async def app_main() -> None:
    schema = await neo4j.TableSchema.from_class(Document, primary_key="filename")
    documents = await neo4j.mount_node_target(
        KG_DB, "Document", schema, primary_key="filename",
    )
    documents.declare_record(
        row=Document(
            filename="overview.md",
            title="Overview",
            summary="An overview of CocoIndex...",
        )
    )


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

Example: Relationship tables (knowledge graph)

@dataclass
class Entity:
    value: str


@dataclass
class RelationshipRow:
    id: str
    predicate: str


@coco.fn
async def kg_app_main() -> None:
    documents = await neo4j.mount_node_target(
        KG_DB, "Document",
        await neo4j.TableSchema.from_class(Document, primary_key="filename"),
        primary_key="filename",
    )
    entities = await neo4j.mount_node_target(
        KG_DB, "Entity",
        await neo4j.TableSchema.from_class(Entity, primary_key="value"),
        primary_key="value",
    )
    relationships = await neo4j.mount_relationship_target(
        KG_DB, "RELATIONSHIP",
        entities, entities,
        await neo4j.TableSchema.from_class(RelationshipRow, primary_key="id"),
        primary_key="id",
    )

    # populate ...
    documents.declare_record(row=Document(filename="overview.md", title="Overview", summary="..."))
    entities.declare_record(row=Entity(value="CocoIndex"))
    entities.declare_record(row=Entity(value="Neo4j"))
    relationships.declare_relationship(
        from_id="CocoIndex",
        to_id="Neo4j",
        record=RelationshipRow(id="rel-1", predicate="writes_to"),
    )


kg_app = coco.App(coco.AppConfig(name="kg_app"), kg_app_main)

The Entity table is declared up-front (via mount_node_target) so its uniqueness constraint is reconciled before any RELATIONSHIP edge MERGEs entity endpoints. The relationship’s three-MERGE pattern (source endpoint → target endpoint → edge) means missing endpoints are auto-created — but it’s good practice to declare them explicitly so deletion-cascade behavior stays predictable.

Deprecated: table and relation APIs

The table and relation APIs are deprecated. They emit a DeprecationWarning and otherwise work as before. Each has a replacement with the same parameters:

One parameter check differs: the relationship APIs require primary_key to match the schema’s primary key, as the node APIs do, where the relation APIs used the schema’s.

The deprecated APIs keep the keys and edge ids of existing tables, and with them three limits the replacements remove:

• A node table and a relationship table share one key space, so a label and a relationship type must have different names. Declaring both under one name fails the update.
• The derived edge id is {from_label}_{from_id}_{to_label}_{to_id}. Two endpoint pairs can share it (the ids 1 and "1", or ids that contain _), and declaring both fails the update.
• A label or relationship type may contain __ or end with _, so two tables’ constraint or index names can coincide.

A label belongs to one table: don’t declare the same label (or relationship type) through both the deprecated and the new API. The two declarations are separate tables, and removing either one deletes every node of the label (or edge of the type).

Switching an existing table

Switching a table from the deprecated API to its replacement keeps its nodes, edges, constraints and indexes. It takes two updates:

1. Add managed_by="user" to each deprecated declaration you switch, and run one update. Nothing in the database changes. From now on, removing these declarations deletes nothing.
2. Replace the deprecated calls with their replacements (and declare_relation with declare_relationship), without managed_by="user", and run one update. The new tables find their constraints and indexes in place, under the same names. Every node and edge of a switched table is written once more, and its vector indexes are rebuilt. An edge with a derived id gets the new id: the edge with the old id between the same endpoints is replaced.

Run the two steps as separate updates, in this order. Without step 1, removing a deprecated declaration destroys its table: every node of the label or edge of the type is deleted, possibly after the new table has written it.

You can switch tables one at a time: a new relationship table accepts node targets of both APIs.