SurrealDB connector

Write to SurrealDB with support for normal and relation (graph edge) tables, typed values, array record IDs, bounded transactions, and HNSW vector indexes with cosine / euclidean / manhattan distances.

Version: v 1.0.14
Last reviewed: Jul 4, 2026

The surrealdb connector provides utilities for writing records to SurrealDB databases, with support for normal tables, relation (graph edge) tables, optional schema enforcement, and vector indexes.

python

from cocoindex.connectors import surrealdb

Dependencies

This connector requires additional dependencies. Install with:

bash

pip install cocoindex[surrealdb]

Connection setup

Create a ConnectionFactory and provide it via a ContextKey. It holds connection parameters and opens one authenticated connection per event loop, shared by all tables and batches. Call await factory.close() at shutdown to close it.

Note

The key name is load-bearing across runs — it’s the stable identity CocoIndex uses to track managed rows. See ContextKey as stable identity before renaming.

python

from cocoindex.connectors import surrealdb
import cocoindex as coco

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

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

As target

The surrealdb connector provides target state APIs for writing records to normal tables and relation tables. CocoIndex tracks what records should exist and automatically handles upserts and deletions.

Writes are sent as typed parameters (datetime, date, Decimal, UUID, bytes, timedelta, lists and dicts arrive as native SurrealDB types, not strings).

Transactions and errors. The changes of each table are applied in chunks of at most 500 actions, each chunk one transaction (Transaction conflict errors are retried with backoff). Atomicity is per chunk, not across tables. Any statement error rolls the chunk back and raises, so the update fails and the affected components are retried on the next run.

Record IDs may be a string, an integer, or a tuple/list, which becomes an array ID (person:['sap', '0001']). Relation endpoints (from_id, to_id) take the same forms.

Ownership. declare_record writes with UPSERT ... MERGE, never CONTENT: fields written by other owners survive, and fields you stop declaring are unset. Use declare_fields when a pipeline contributes only some fields of a record that another pipeline owns. Relation records are single-owner and are replaced whole.

DDL. Table and field definitions use OVERWRITE, so they are idempotent. With managed_by="user" CocoIndex never runs DDL on tables (indexes declared with declare_vector_index are still created).

Declaring target states

Normal tables (parent state)

Declares a table as a target state. Returns a TableTarget for declaring records.

python

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

Parameters:

db — A ContextKey[surrealdb.ConnectionFactory] for the SurrealDB connection.
table_name — Name of the table.
table_schema — Optional schema definition (see Table Schema). When provided, the table is SCHEMAFULL; when omitted, the table is SCHEMALESS.
managed_by — Whether CocoIndex manages the table lifecycle ("system") or assumes it exists ("user").

Returns: A pending TableTarget. Use await surrealdb.mount_table_target(SURREAL_DB, ...) to get a resolved target.

Records (child states)

Once a TableTarget is resolved, declare records to be upserted:

python

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

Parameters:

row — A row object (dict, dataclass, NamedTuple, or Pydantic model). Must include an id field.

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

python

def TableTarget.declare_fields(
    self,
    *,
    id: Any,
    group: str,
    fields: dict[str, Any],
) -> None

Declares the fields one owner (group) contributes to a record. It runs UPDATE $id MERGE {fields} and touches nothing else. When the declaration goes away, or a field is dropped from it, those fields are unset. If the record does not exist, the write raises and is retried on the next run, so declare the base record in the same or another pipeline.

Relation tables (parent state)

Declares a relation (graph edge) table. Returns a RelationTarget for declaring relation records.

python

def declare_relation_target(
    db: ContextKey,
    table_name: str,
    from_table: TableTarget | Collection[TableTarget],
    to_table: TableTarget | Collection[TableTarget],
    table_schema: TableSchema[RowT] | None = None,
    *,
    managed_by: Literal["system", "user"] = "system",
) -> RelationTarget[RowT, coco.PendingS]

Parameters:

db — A ContextKey[surrealdb.ConnectionFactory] for the SurrealDB connection.
table_name — Name of the relation table.
from_table — Source table(s). Pass a single TableTarget or a collection for polymorphic relations.
to_table — Target table(s). Same rules as from_table.
table_schema — Optional schema. The schema does not require an id field (unlike normal tables).
managed_by — Whether CocoIndex manages the table lifecycle.

Returns: A pending RelationTarget. Use await surrealdb.mount_relation_target(SURREAL_DB, ...) to get a resolved target.

Relations (child states)

Once a RelationTarget is resolved, declare relation records:

python

def RelationTarget.declare_relation(
    self,
    *,
    from_id: Any,
    to_id: Any,
    record: RowT | None = None,
    from_table: TableTarget | None = None,
    to_table: TableTarget | None = None,
) -> None

Parameters:

from_id — ID of the source record.
to_id — ID of the target record.
record — Optional data fields for the relation. The id field is optional: when absent, the record id is auto-derived from the endpoints as "{from_table}_{from_id}_{to_table}_{to_id}".
from_table / to_table — Required when the relation was declared with multiple (polymorphic) source/target tables.

Vector indexes (attachment)

Declare a vector index on a field of the table. CocoIndex tracks the index spec and automatically creates, recreates, or drops the index as needed.

python

def TableTarget.declare_vector_index(
    self,
    *,
    name: str | None = None,
    field: str,
    metric: Literal["cosine", "euclidean", "manhattan"] = "cosine",
    method: Literal["hnsw"] = "hnsw",
    dimension: int | None = None,
    vector_type: Literal["f32", "f64", "i16", "i32", "i64"] = "f32",
) -> None

Parameters:

name — Index name (defaults to idx_{table}__{field}).
field — Field to index (must be a vector/array field).
metric — Distance metric: "cosine", "euclidean", or "manhattan".
method — Index method: "hnsw" (SurrealDB 3 removed MTREE).
dimension — Vector dimension (required).
vector_type — Vector element type: "f32", "f64", "i16", "i32", or "i64".

Table schema: from Python class

Define the table structure using a Python class (dataclass, NamedTuple, or Pydantic model):

python

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

Parameters:

record_type — A record type whose fields define table columns. For normal tables, must include an id field. For relation tables, id is optional.
column_overrides — Optional per-column overrides for type mapping or vector configuration.

Example:

python

@dataclass
class Product:
    id: str
    name: str
    price: float
    embedding: Annotated[NDArray, embedder]

schema = await surrealdb.TableSchema.from_class(Product)

Python types are automatically mapped to SurrealDB types:

Python Type	SurrealDB Type
`bool`	`bool`
`int`	`int`
`float`	`float`
`decimal.Decimal`	`decimal`
`str`	`string`
`bytes`	`bytes`
`uuid.UUID`	`uuid`
`datetime.datetime`	`datetime`
`datetime.date`	`datetime`
`datetime.time`	`datetime`
`datetime.timedelta`	`duration`
`list`, `dict`, nested structs	`object`
`NDArray` (with vector schema)	`array<float, N>`

SurrealType

Use SurrealType to override the default type mapping:

python

from typing import Annotated
from cocoindex.connectors.surrealdb import SurrealType

@dataclass
class MyRow:
    id: str
    value: Annotated[float, SurrealType("decimal")]

Or via column_overrides:

python

schema = await surrealdb.TableSchema.from_class(
    MyRow,
    column_overrides={"value": surrealdb.SurrealType("decimal")},
)

VectorSchemaProvider

For NDArray fields, a VectorSchemaProvider annotation specifies the vector dimension and dtype. See Vector Schema for the full list of annotation options.

Table schema: explicit column definitions

Define columns directly using ColumnDef:

python

def TableSchema.__init__(
    self,
    columns: dict[str, ColumnDef],
    *,
    row_type: type[RowT] | None = None,
) -> None

Example:

python

schema = surrealdb.TableSchema(
    {
        "id": surrealdb.ColumnDef(type="string", nullable=False),
        "name": surrealdb.ColumnDef(type="string", nullable=False),
        "price": surrealdb.ColumnDef(type="float"),
    },
)

Example: Normal tables

python

import cocoindex as coco
from cocoindex.connectors import surrealdb

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

@dataclass
class Product:
    id: str
    name: str
    price: float
    embedding: Annotated[NDArray, embedder]

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

@coco.fn
async def app_main() -> None:
    # Declare table target state
    table = await surrealdb.mount_table_target(
        SURREAL_DB,
        "products",
        await surrealdb.TableSchema.from_class(Product),
    )

    # Declare records
    for product in products:
        table.declare_record(row=product)

    # Declare a vector index
    table.declare_vector_index(
        field="embedding",
        metric="cosine",
        method="hnsw",
        dimension=384,
    )

Example: Relation tables

python

@dataclass
class Person:
    id: str
    name: str

@dataclass
class Post:
    id: str
    title: str

@coco.fn
async def app_main() -> None:
    person_schema = await surrealdb.TableSchema.from_class(Person)
    person_target = await surrealdb.mount_table_target(SURREAL_DB, "person", person_schema)
    for p in persons:
        person_target.declare_record(row=p)

    post_schema = await surrealdb.TableSchema.from_class(Post)
    post_target = await surrealdb.mount_table_target(SURREAL_DB, "post", post_schema)
    for p in posts:
        post_target.declare_record(row=p)

    # Declare a relation table (schemaless, no id needed)
    likes_target = await surrealdb.mount_relation_target(
        SURREAL_DB,
        "likes",
        from_table=person_target,
        to_table=post_target,
    )

    # Declare relations — id is auto-derived from endpoints
    for like in likes:
        likes_target.declare_relation(
            from_id=like["person_id"],
            to_id=like["post_id"],
        )