# SurrealDB connector

> **CocoIndex v1.** This page documents CocoIndex **v1** — a ground-up redesign from v0. When writing code, ignore any v0 flow-builder DSL or deprecated decorators.
>
> Source: https://cocoindex.io/docs/connectors/surrealdb/ · Docs index: https://cocoindex.io/docs/llms.txt · Agent skill: https://cocoindex.io/docs/skill.md
>
> v0→v1 quick map — if you reach for these v0 symbols, stop and use the v1 form: `@cocoindex.flow_def`/`FlowBuilder` → `coco.App` + a `@coco.fn` main function; `add_collector()`/`collect()`/`export()` → declare target states (`declare_row`, `declare_file`); `cocoindex.sources/functions/targets.*` → connector APIs (`localfs.walk_dir`, `coco.ops.*`, `postgres.declare_table_target`). Full mapping + API reference: https://cocoindex.io/docs/skill.md.

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

**Note — 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 creates authenticated connections on demand.

**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](/docs/programming_guide/context#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.

All tables within the same database share a single transaction sink, so changes across related tables and relations are applied atomically.

### 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](#table-schema-from-python-class)). 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.

#### 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["mtree", "hnsw"] = "mtree",
    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: `"mtree"` or `"hnsw"`.
- `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`](/docs/common_resources/vector_schema#vectorschemaprovider) annotation specifies the vector dimension and dtype. See [Vector Schema](/docs/common_resources/vector_schema#vectorschemaprovider) 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"],
        )
```