VectorIndex

High-level interface to Chassis vector index.

This class provides a Pythonic wrapper around the Chassis FFI layer, handling memory management, error handling, and type conversions.

Thread Safety

Every method is safe from any thread. Searches run concurrently; add(), delete() and flush() run one at a time and searches wait for them. Don't call close() while other threads still use the index. An index opened with read_only=True runs one search at a time.

Readers in other processes

Any number of processes can open the file with read_only=True while one process writes it. Each search sees the writer's last flush.

Example

index = VectorIndex("vectors.chassis", dimensions=128)

Add vectors

vectors = np.random.rand(1000, 128).astype(np.float32) for vec in vectors: ... index.add(vec) index.flush()

Search

query = np.random.rand(128).astype(np.float32) results = index.search(query, k=10) for result in results: ... print(f"ID: {result.id}, Distance: {result.distance}")

Source code in chassisdb/index.py

class VectorIndex:
    """High-level interface to Chassis vector index.

    This class provides a Pythonic wrapper around the Chassis FFI layer,
    handling memory management, error handling, and type conversions.

    Thread Safety:
        Every method is safe from any thread. Searches run concurrently;
        add(), delete() and flush() run one at a time and searches wait for
        them. Don't call close() while other threads still use the index.
        An index opened with read_only=True runs one search at a time.

    Readers in other processes:
        Any number of processes can open the file with read_only=True while
        one process writes it. Each search sees the writer's last flush.

    Example:
        >>> index = VectorIndex("vectors.chassis", dimensions=128)
        >>>
        >>> # Add vectors
        >>> vectors = np.random.rand(1000, 128).astype(np.float32)
        >>> for vec in vectors:
        ...     index.add(vec)
        >>> index.flush()
        >>>
        >>> # Search
        >>> query = np.random.rand(128).astype(np.float32)
        >>> results = index.search(query, k=10)
        >>> for result in results:
        ...     print(f"ID: {result.id}, Distance: {result.distance}")
    """

    def __init__(
        self,
        path: str | Path,
        dimensions: int,
        options: IndexOptions | None = None,
        read_only: bool = False,
    ):
        """Open or create a vector index.

        Args:
            path: Path to the index file
            dimensions: Number of dimensions per vector
            options: Optional HNSW configuration. If None, uses defaults.
            read_only: Open an existing index to search it while another
                process writes it. Takes no lock; add(), delete() and
                flush() raise ChassisError. The index keeps the metric it
                was created with; explicit options naming another raise.

        Raises:
            IndexNotFoundError: If read_only and there is no index at path, or
                the path's directory doesn't exist
            IndexLockedError: If another writer has the index open
            DimensionMismatchError: If the index holds vectors of other
                dimensions
            OptionsMismatchError: If the index was created with another
                metric or precision
            NotAnIndexError: If the file isn't a Chassis index
            InvalidArgumentError: If dimensions or an option is out of range
            TypeError: If an argument has the wrong type
            ChassisError: For other errors; every class above is one too
        """
        # Set before anything can raise: __del__ calls close(), which reads them.
        self._ptr: ctypes._Pointer[_ffi.ChassisIndex] | None = None
        self._closed = False
        self._read_only = read_only
        self._path = Path(path)
        self._dimensions = _check_dimensions(dimensions)
        self._options = _check_options(options)
        explicit = options is not None
        options = self._options

        # Encode path to UTF-8 bytes
        path_bytes = str(self._path).encode("utf-8")

        # Open index with options
        if read_only:
            ptr = _ffi._lib.chassis_open_reader(
                path_bytes,
                dimensions,
                self._options.max_connections,
                self._options.ef_search,
            )
        elif not explicit:
            # Use default options
            ptr = _ffi._lib.chassis_open(path_bytes, dimensions)
        else:
            # Use custom options
            ptr = _ffi._lib.chassis_open_with_precision(
                path_bytes,
                dimensions,
                options.max_connections,
                options.ef_construction,
                options.ef_search,
                _METRICS[options.metric],
                _PRECISIONS[options.precision],
            )

        if not ptr:
            raise _ffi.last_error("The index couldn't be opened, and Chassis didn't say why")

        self._ptr = ptr
        if self._options.huge_pages:
            _ffi._lib.chassis_use_huge_pages(ptr)
        if self._options.warm:
            _ffi._lib.chassis_warm(ptr)
        # A reader takes the file's metric and precision; say so rather than use others.
        metric, precision = self.metric, self.precision
        if explicit and options.metric != metric:
            self.close()
            raise OptionsMismatchError(
                f"The index at {self._path} was created with {metric} distance, not "
                f"{options.metric}\nhelp: open it with metric={metric!r}, or leave the metric "
                "out: a read_only index takes it from the file"
            )
        if explicit and options.precision != precision:
            self.close()
            raise OptionsMismatchError(
                f"The index at {self._path} keeps its vectors in {precision} precision, not "
                f"{options.precision}\nhelp: open it with precision={precision!r}, or leave the "
                "precision out: a read_only index takes it from the file"
            )
        # `options` reports what the index really uses.
        self._options = dataclasses.replace(self._options, metric=metric, precision=precision)

    @property
    def metric(self) -> str:
        """The distance metric the index was created with: "euclidean" or "cosine"."""
        self._check_closed()
        return {v: k for k, v in _METRICS.items()}[_ffi._lib.chassis_metric(self._ptr)]

    @property
    def precision(self) -> str:
        """What the index keeps of each component of a vector: "full" or "half"."""
        self._check_closed()
        return {v: k for k, v in _PRECISIONS.items()}[_ffi._lib.chassis_precision(self._ptr)]

    def __del__(self) -> None:
        """Clean up resources when index is garbage collected."""
        self.close()

    def __enter__(self) -> "VectorIndex":
        """Context manager entry."""
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> Literal[False]:
        """Flush, unless an exception is leaving the block, then close."""
        try:
            if exc_type is None and not self._read_only and not self._closed:
                self.flush()
        finally:
            self.close()
        return False

    def close(self) -> None:
        """Close the index and free resources.

        Adds and deletes since the last flush() are discarded, as after a
        crash. Leaving a with block flushes first, unless an exception is
        leaving it. Garbage collection closes without flushing. It's safe to
        call multiple times.
        """
        if not self._closed and self._ptr:
            _ffi._lib.chassis_free(self._ptr)
            self._ptr = None
            self._closed = True

    def _check_closed(self) -> None:
        """Check if index is closed and raise error if so."""
        if self._closed or not self._ptr:
            raise ChassisError(
                f"The index at {self._path} is closed\nhelp: open it again with "
                "VectorIndex(path, dimensions), or use it inside its with block"
            )

    def _check_writable(self, action: str) -> None:
        self._check_closed()
        if self._read_only:
            raise ReadOnlyError(
                f"Can't {action}: the index at {self._path} was opened with read_only=True, which "
                "only searches\nhelp: open it without read_only to write it; an index has one "
                "writer at a time, and read_only indexes can search while it writes"
            )

    def add(
        self,
        vector: Sequence[float] | npt.NDArray[np.float32],
        id: int | None = None,
    ) -> int:
        """Add a vector to the index.

        Args:
            vector: Vector to add (must match index dimensions)
                Can be a list, tuple, numpy array, or any sequence of floats
            id: Id to store the vector under, from 0 to 2**64 - 2. Search
                results report it. Default: one past the largest id used so far.

        Returns:
            The vector's id

        Raises:
            ChassisError: If index is closed
            DimensionMismatchError: If vector dimensions don't match
            ValueError: If id is out of range
            ChassisError: If a vector with this id already exists (delete it
                first to replace it), or for other errors

        Note:
            This method does NOT guarantee durability. Call flush() to
            ensure data is written to disk.

        Thread Safety:
            Safe from any thread; writes run one at a time.
        """
        self._check_writable("add")
        vector = _as_vector(vector, self._dimensions, "vector")
        vector_ptr = vector.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
        if id is not None:
            id = _check_id(id)
            if _ffi._lib.chassis_add_with_id(self._ptr, id, vector_ptr, len(vector)) != 0:
                raise _ffi.last_error("The vector couldn't be added")
            return id
        vector_id = _ffi._lib.chassis_add(self._ptr, vector_ptr, len(vector))
        if vector_id == 2**64 - 1:
            raise _ffi.last_error("The vector couldn't be added")
        return int(vector_id)

    def add_batch(
        self,
        vectors: Sequence[Sequence[float]] | npt.NDArray[np.float32],
        ids: Iterable[int] | npt.NDArray[np.uint64] | None = None,
    ) -> npt.NDArray[np.uint64]:
        """Add many vectors at once, linking them on every core.

        Much faster than calling add() in a loop for large batches; the
        graph then depends on thread timing. All or nothing: if any vector
        or id is rejected, none of the batch is added.

        Args:
            vectors: A (count, dimensions) array, or a sequence of vectors
            ids: Optional ids, one per vector, as for add(). Default: each
                vector gets the id add() would give it.

        Returns:
            The vectors' ids

        Raises:
            DimensionMismatchError: If the rows don't match the index dimensions
            ValueError: If ids are out of range or their count differs
            ChassisError: If an id repeats or already exists, or for other errors
        """
        self._check_writable("add")
        vectors = _as_batch(vectors, self._dimensions)
        count = len(vectors)
        vectors_ptr = vectors.ctypes.data_as(ctypes.POINTER(ctypes.c_float))

        if ids is None:
            out_ids = np.zeros(count, dtype=np.uint64)
            out_ptr = out_ids.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
            added = _ffi._lib.chassis_add_batch(
                self._ptr, vectors_ptr, count, self._dimensions, out_ptr
            )
            if added != count:
                raise _ffi.last_error("The batch couldn't be added")
            return out_ids

        ids = [_check_id(i) for i in (ids.tolist() if isinstance(ids, np.ndarray) else ids)]
        if len(ids) != count:
            raise InvalidArgumentError(
                f"{len(ids)} ids were given for {count} vectors\nhelp: give one id per vector, "
                "or leave ids out to have them chosen"
            )
        id_array = np.array(ids, dtype=np.uint64)
        ids_ptr = id_array.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
        if (
            _ffi._lib.chassis_add_batch_with_ids(
                self._ptr, ids_ptr, vectors_ptr, count, self._dimensions
            )
            != 0
        ):
            raise _ffi.last_error("The batch couldn't be added")
        return id_array

    def delete(self, id: int) -> bool:
        """Delete the vector with this id.

        Search stops returning it immediately. The delete is durable after the
        next flush(); a crash before then rolls it back.

        Returns:
            True if it was deleted, False if no vector has this id

        Raises:
            ChassisError: If index is closed or the delete fails
            ValueError: If id is out of range

        Thread Safety:
            Safe from any thread; writes run one at a time.
        """
        self._check_writable("delete")
        result: int = _ffi._lib.chassis_delete(self._ptr, _check_id(id))
        if result < 0:
            raise _ffi.last_error("The vector couldn't be deleted")
        return result == 1

    def get(self, id: int) -> npt.NDArray[np.float32] | None:
        """The vector with this id, or None if no live vector has it.

        It comes back as the index keeps it: scaled to length 1 in a cosine
        index, and rounded in half precision.

        Raises:
            ChassisError: If index is closed or the file can't be read
            ValueError: If id is out of range

        Thread Safety:
            Safe from any thread. A read-only index sees the writer's last
            flush.
        """
        self._check_closed()
        out = np.empty(self._dimensions, dtype=np.float32)
        out_ptr = out.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
        result: int = _ffi._lib.chassis_get(self._ptr, _check_id(id), out_ptr, self._dimensions)
        if result < 0:
            raise _ffi.last_error("The vector couldn't be read")
        return out if result == 1 else None

    def __contains__(self, id: object) -> bool:
        """Whether a live vector has this id: ``id in index``."""
        self._check_closed()
        try:
            id = _check_id(id)
        except InvalidArgumentError:
            return False  # no vector has an id out of range
        result: int = _ffi._lib.chassis_contains(self._ptr, id)
        if result < 0:
            raise _ffi.last_error("The id couldn't be looked up")
        return result == 1

    def ids(self) -> npt.NDArray[np.uint64]:
        """The ids of the live vectors, in the order they were added.

        Thread Safety:
            Safe from any thread. A read-only index sees the writer's last
            flush.
        """
        self._check_closed()
        out = np.empty(len(self), dtype=np.uint64)
        while True:
            out_ptr = out.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
            count = _ffi._lib.chassis_ids(self._ptr, out_ptr, len(out))
            if count == 0 and _ffi._lib.chassis_last_error_code() != 0:
                raise _ffi.last_error("The ids couldn't be read")
            if count <= len(out):
                return out[:count]
            # A writer in another process may flush more before the next call.
            out = np.empty(count, dtype=np.uint64)

    def search(
        self,
        query: Sequence[float] | npt.NDArray[np.float32],
        k: int = 10,
        allowed: Iterable[int] | npt.NDArray[np.uint64] | None = None,
        excluded: Iterable[int] | npt.NDArray[np.uint64] | None = None,
    ) -> list[SearchResult]:
        """Search for k nearest neighbors.

        Args:
            query: Query vector (must match index dimensions)
            k: Number of nearest neighbors to return (default: 10)
            allowed: If given, only these ids can be returned. When walking
                the graph would cost more, as when few vectors match, every
                vector is checked instead and the results are exact. A
                numpy array of ids in ascending order costs the least; don't
                pass every id just to keep stray vectors out.
            excluded: If given, these ids are never returned, as for "more
                like this" leaving out the vector it starts from.

        Returns:
            List of SearchResult objects, sorted by distance (ascending)

        Raises:
            ChassisError: If index is closed
            DimensionMismatchError: If query dimensions don't match
            ValueError: If k < 1
            ChassisError: For other errors

        Thread Safety:
            Safe from any thread. Runs concurrently with other searches and
            waits while a write runs.
        """
        self._check_closed()
        k = _check_int("k", k)
        if k < 1:
            raise InvalidArgumentError(
                f"k is {k}, but has to be at least 1\nhelp: pass how many results to return, "
                "such as k=10"
            )
        query = _as_vector(query, self._dimensions, "query")

        # Allocate output buffers
        out_ids = np.zeros(k, dtype=np.uint64)
        out_dists = np.zeros(k, dtype=np.float32)

        # Call FFI
        query_ptr = query.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
        ids_ptr = out_ids.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
        dists_ptr = out_dists.ctypes.data_as(ctypes.POINTER(ctypes.c_float))

        if excluded is not None:
            excluded = _id_array("excluded", excluded)
        if allowed is not None:
            allowed = _id_array("allowed", allowed)
            if excluded is not None:
                allowed = np.setdiff1d(allowed, excluded)

        if allowed is not None:
            allowed_ptr = allowed.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
            count = _ffi._lib.chassis_search_filtered(
                self._ptr, query_ptr, len(query), k, allowed_ptr, len(allowed), ids_ptr, dists_ptr
            )
        elif excluded is not None:
            excluded_ptr = excluded.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
            count = _ffi._lib.chassis_search_excluding(
                self._ptr, query_ptr, len(query), k, excluded_ptr, len(excluded), ids_ptr, dists_ptr
            )
        else:
            count = _ffi._lib.chassis_search(
                self._ptr, query_ptr, len(query), k, ids_ptr, dists_ptr
            )

        # 0 results is an error only if one was set: an empty index has none either.
        if count == 0 and _ffi._lib.chassis_last_error_code() != 0:
            raise _ffi.last_error("The search failed")

        # Convert to SearchResult objects
        results = [
            SearchResult(id=int(out_ids[i]), distance=float(out_dists[i])) for i in range(count)
        ]

        return results

    def search_batch(
        self,
        queries: Sequence[Sequence[float]] | npt.NDArray[np.float32],
        k: int = 10,
    ) -> tuple[npt.NDArray[np.uint64], npt.NDArray[np.float32]]:
        """Search for the k nearest neighbors of many queries, in one call.

        Faster than calling search once per query: the queries cross from
        Python into the library together.

        Args:
            queries: A (count, dimensions) array, one query per row
            k: Number of nearest neighbors per query (default: 10)

        Returns:
            (ids, distances), two (count, k) arrays. Row i holds query i's
            results, nearest first; a row with fewer than k results is filled
            out with id 2**64 - 1, which no vector has, and distance inf.

        Raises:
            ChassisError: If index is closed
            DimensionMismatchError: If the queries aren't rows of the index's
                dimensions; the message names a query that is wrong
            InvalidArgumentError: If k < 1, or a query has a component that
                isn't a finite number
            ChassisError: For other errors

        Thread Safety:
            Safe from any thread. Runs concurrently with other searches, and a
            write waits for the whole batch.
        """
        self._check_closed()
        k = _check_int("k", k)
        if k < 1:
            raise InvalidArgumentError(
                f"k is {k}, but has to be at least 1\nhelp: pass how many results to return per "
                "query, such as k=10"
            )
        queries = _as_batch(queries, self._dimensions, "query", "search")
        count = len(queries)
        out_ids = np.empty((count, k), dtype=np.uint64)
        out_dists = np.empty((count, k), dtype=np.float32)
        if count == 0:
            return out_ids, out_dists

        done = _ffi._lib.chassis_search_batch(
            self._ptr,
            queries.ctypes.data_as(ctypes.POINTER(ctypes.c_float)),
            count,
            self._dimensions,
            k,
            out_ids.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64)),
            out_dists.ctypes.data_as(ctypes.POINTER(ctypes.c_float)),
        )
        if done != count:
            raise _ffi.last_error("The search failed")
        return out_ids, out_dists

    def flush(self) -> None:
        """Flush all changes to disk.

        This method ensures durability by writing all pending changes to disk.
        It's expensive (1-50ms) so batch multiple add() calls before flushing.

        Raises:
            ChassisError: If flush fails

        Thread Safety:
            Safe from any thread; writes run one at a time.
        """
        self._check_writable("flush")
        if _ffi._lib.chassis_flush(self._ptr) != 0:
            raise _ffi.last_error("The flush failed")

    def warm(self) -> None:
        """Start reading the whole index into memory on another thread.

        Doesn't wait for the reading: searches go on meanwhile. While it
        is under way, calling again does nothing. Opening with
        ``IndexOptions(warm=True)`` does the same from the start.

        Raises:
            ChassisError: If the index can't be used any more
        """
        self._check_closed()

        if _ffi._lib.chassis_warm(self._ptr) != 0:
            raise _ffi.last_error("The index couldn't be read in")

    def compact(self) -> None:
        """Rewrite the index without its deleted vectors, with a newly built graph.

        Reclaims the space of deleted vectors and replaces the index file with
        the copy. Ids don't change. Like flush(), it makes every add and
        delete so far durable. Takes as long as building the index, on every
        core, and needs free disk for a second copy of the live vectors.
        Readers in other processes keep searching and move to the new file by
        themselves.

        Raises:
            ChassisError: If the copy can't be written or, on Windows, another
                process has the index open. The index is left as it was.
        """
        self._check_writable("compact")
        if _ffi._lib.chassis_compact(self._ptr) != 0:
            raise _ffi.last_error("The compaction failed")

    def __len__(self) -> int:
        """Get the number of vectors in the index.

        Returns:
            Number of vectors

        Thread Safety:
            Safe from any thread.
        """
        self._check_closed()
        return int(_ffi._lib.chassis_len(self._ptr))

    def deleted_count(self) -> int:
        """Deleted vectors the file still holds, until compact() removes them.

        Thread Safety:
            Safe from any thread.
        """
        self._check_closed()
        return int(_ffi._lib.chassis_deleted_count(self._ptr))

    @property
    def user_version(self) -> int:
        """Your application's number for the index, 0 until you set one.

        Like SQLite's user_version. The next flush() commits it with the
        data, so after a crash it says which flush the index holds: keep
        the same number in your own store, in the transaction the flush
        sits in, and compare the two on open. Until that flush, a crash
        leaves the number flushed before. A read_only index reports the one
        in its newest snapshot.

        Raises:
            ReadOnlyError: On setting it in a read_only index
            InvalidArgumentError: If set below 0 or to 2**64 or more
            TypeError: If set to something other than an int
        """
        self._check_closed()
        return int(_ffi._lib.chassis_user_version(self._ptr))

    @user_version.setter
    def user_version(self, version: int) -> None:
        self._check_writable("set user_version")
        version = _check_int("user_version", version)
        if not 0 <= version < 2**64:
            raise InvalidArgumentError(
                f"user_version is {version}, but has to be from 0 to 2**64 - 1\nhelp: pass a "
                "count that grows with each flush, such as 1, 2, 3"
            )
        if _ffi._lib.chassis_set_user_version(self._ptr, version) != 0:
            raise _ffi.last_error("Setting user_version failed")

    def is_empty(self) -> bool:
        """Check if the index is empty.

        Returns:
            True if empty, False otherwise

        Thread Safety:
            Safe from any thread.
        """
        self._check_closed()
        return bool(_ffi._lib.chassis_is_empty(self._ptr))

    @property
    def dimensions(self) -> int:
        """Get the dimensionality of vectors in this index.

        Returns:
            Number of dimensions

        Thread Safety:
            Safe from any thread.
        """
        self._check_closed()
        return int(_ffi._lib.chassis_dimensions(self._ptr))

    @property
    def path(self) -> Path:
        """Get the path to the index file."""
        return self._path

    @property
    def options(self) -> IndexOptions:
        """Get the index configuration options."""
        return self._options

    def __repr__(self) -> str:
        status = "closed" if self._closed else "open"
        return (
            f"VectorIndex(path={self._path}, "
            f"dimensions={self._dimensions}, "
            f"len={len(self) if not self._closed else '?'}, "
            f"status={status})"
        )

dimensions property

Get the dimensionality of vectors in this index.

Returns:

Type	Description
int	Number of dimensions

Thread Safety

Safe from any thread.

metric property

The distance metric the index was created with: "euclidean" or "cosine".

precision property

What the index keeps of each component of a vector: "full" or "half".

path property

Get the path to the index file.

options property

Get the index configuration options.

__init__(path, dimensions, options=None, read_only=False)

Open or create a vector index.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the index file	required
dimensions	int	Number of dimensions per vector	required
options	IndexOptions | None	Optional HNSW configuration. If None, uses defaults.	None
read_only	bool	Open an existing index to search it while another process writes it. Takes no lock; add(), delete() and flush() raise ChassisError. The index keeps the metric it was created with; explicit options naming another raise.	False

Raises:

Type	Description
IndexNotFoundError	If read_only and there is no index at path, or the path's directory doesn't exist
IndexLockedError	If another writer has the index open
DimensionMismatchError	If the index holds vectors of other dimensions
OptionsMismatchError	If the index was created with another metric or precision
NotAnIndexError	If the file isn't a Chassis index
InvalidArgumentError	If dimensions or an option is out of range
TypeError	If an argument has the wrong type
ChassisError	For other errors; every class above is one too

Source code in chassisdb/index.py

def __init__(
    self,
    path: str | Path,
    dimensions: int,
    options: IndexOptions | None = None,
    read_only: bool = False,
):
    """Open or create a vector index.

    Args:
        path: Path to the index file
        dimensions: Number of dimensions per vector
        options: Optional HNSW configuration. If None, uses defaults.
        read_only: Open an existing index to search it while another
            process writes it. Takes no lock; add(), delete() and
            flush() raise ChassisError. The index keeps the metric it
            was created with; explicit options naming another raise.

    Raises:
        IndexNotFoundError: If read_only and there is no index at path, or
            the path's directory doesn't exist
        IndexLockedError: If another writer has the index open
        DimensionMismatchError: If the index holds vectors of other
            dimensions
        OptionsMismatchError: If the index was created with another
            metric or precision
        NotAnIndexError: If the file isn't a Chassis index
        InvalidArgumentError: If dimensions or an option is out of range
        TypeError: If an argument has the wrong type
        ChassisError: For other errors; every class above is one too
    """
    # Set before anything can raise: __del__ calls close(), which reads them.
    self._ptr: ctypes._Pointer[_ffi.ChassisIndex] | None = None
    self._closed = False
    self._read_only = read_only
    self._path = Path(path)
    self._dimensions = _check_dimensions(dimensions)
    self._options = _check_options(options)
    explicit = options is not None
    options = self._options

    # Encode path to UTF-8 bytes
    path_bytes = str(self._path).encode("utf-8")

    # Open index with options
    if read_only:
        ptr = _ffi._lib.chassis_open_reader(
            path_bytes,
            dimensions,
            self._options.max_connections,
            self._options.ef_search,
        )
    elif not explicit:
        # Use default options
        ptr = _ffi._lib.chassis_open(path_bytes, dimensions)
    else:
        # Use custom options
        ptr = _ffi._lib.chassis_open_with_precision(
            path_bytes,
            dimensions,
            options.max_connections,
            options.ef_construction,
            options.ef_search,
            _METRICS[options.metric],
            _PRECISIONS[options.precision],
        )

    if not ptr:
        raise _ffi.last_error("The index couldn't be opened, and Chassis didn't say why")

    self._ptr = ptr
    if self._options.huge_pages:
        _ffi._lib.chassis_use_huge_pages(ptr)
    if self._options.warm:
        _ffi._lib.chassis_warm(ptr)
    # A reader takes the file's metric and precision; say so rather than use others.
    metric, precision = self.metric, self.precision
    if explicit and options.metric != metric:
        self.close()
        raise OptionsMismatchError(
            f"The index at {self._path} was created with {metric} distance, not "
            f"{options.metric}\nhelp: open it with metric={metric!r}, or leave the metric "
            "out: a read_only index takes it from the file"
        )
    if explicit and options.precision != precision:
        self.close()
        raise OptionsMismatchError(
            f"The index at {self._path} keeps its vectors in {precision} precision, not "
            f"{options.precision}\nhelp: open it with precision={precision!r}, or leave the "
            "precision out: a read_only index takes it from the file"
        )
    # `options` reports what the index really uses.
    self._options = dataclasses.replace(self._options, metric=metric, precision=precision)

add(vector, id=None)

Add a vector to the index.

Parameters:

Name	Type	Description	Default
vector	Sequence[float] | NDArray[float32]	Vector to add (must match index dimensions) Can be a list, tuple, numpy array, or any sequence of floats	required
id	int | None	Id to store the vector under, from 0 to 2**64 - 2. Search results report it. Default: one past the largest id used so far.	None

Returns:

Type	Description
int	The vector's id

Raises:

Type	Description
ChassisError	If index is closed
DimensionMismatchError	If vector dimensions don't match
ValueError	If id is out of range
ChassisError	If a vector with this id already exists (delete it first to replace it), or for other errors

Note

This method does NOT guarantee durability. Call flush() to ensure data is written to disk.

Thread Safety

Safe from any thread; writes run one at a time.

Source code in chassisdb/index.py

def add(
    self,
    vector: Sequence[float] | npt.NDArray[np.float32],
    id: int | None = None,
) -> int:
    """Add a vector to the index.

    Args:
        vector: Vector to add (must match index dimensions)
            Can be a list, tuple, numpy array, or any sequence of floats
        id: Id to store the vector under, from 0 to 2**64 - 2. Search
            results report it. Default: one past the largest id used so far.

    Returns:
        The vector's id

    Raises:
        ChassisError: If index is closed
        DimensionMismatchError: If vector dimensions don't match
        ValueError: If id is out of range
        ChassisError: If a vector with this id already exists (delete it
            first to replace it), or for other errors

    Note:
        This method does NOT guarantee durability. Call flush() to
        ensure data is written to disk.

    Thread Safety:
        Safe from any thread; writes run one at a time.
    """
    self._check_writable("add")
    vector = _as_vector(vector, self._dimensions, "vector")
    vector_ptr = vector.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
    if id is not None:
        id = _check_id(id)
        if _ffi._lib.chassis_add_with_id(self._ptr, id, vector_ptr, len(vector)) != 0:
            raise _ffi.last_error("The vector couldn't be added")
        return id
    vector_id = _ffi._lib.chassis_add(self._ptr, vector_ptr, len(vector))
    if vector_id == 2**64 - 1:
        raise _ffi.last_error("The vector couldn't be added")
    return int(vector_id)

add_batch(vectors, ids=None)

Add many vectors at once, linking them on every core.

Much faster than calling add() in a loop for large batches; the graph then depends on thread timing. All or nothing: if any vector or id is rejected, none of the batch is added.

Parameters:

Name	Type	Description	Default
vectors	Sequence[Sequence[float]] | NDArray[float32]	A (count, dimensions) array, or a sequence of vectors	required
ids	Iterable[int] | NDArray[uint64] | None	Optional ids, one per vector, as for add(). Default: each vector gets the id add() would give it.	None

Returns:

Type	Description
NDArray[uint64]	The vectors' ids

Raises:

Type	Description
DimensionMismatchError	If the rows don't match the index dimensions
ValueError	If ids are out of range or their count differs
ChassisError	If an id repeats or already exists, or for other errors

Source code in chassisdb/index.py

def add_batch(
    self,
    vectors: Sequence[Sequence[float]] | npt.NDArray[np.float32],
    ids: Iterable[int] | npt.NDArray[np.uint64] | None = None,
) -> npt.NDArray[np.uint64]:
    """Add many vectors at once, linking them on every core.

    Much faster than calling add() in a loop for large batches; the
    graph then depends on thread timing. All or nothing: if any vector
    or id is rejected, none of the batch is added.

    Args:
        vectors: A (count, dimensions) array, or a sequence of vectors
        ids: Optional ids, one per vector, as for add(). Default: each
            vector gets the id add() would give it.

    Returns:
        The vectors' ids

    Raises:
        DimensionMismatchError: If the rows don't match the index dimensions
        ValueError: If ids are out of range or their count differs
        ChassisError: If an id repeats or already exists, or for other errors
    """
    self._check_writable("add")
    vectors = _as_batch(vectors, self._dimensions)
    count = len(vectors)
    vectors_ptr = vectors.ctypes.data_as(ctypes.POINTER(ctypes.c_float))

    if ids is None:
        out_ids = np.zeros(count, dtype=np.uint64)
        out_ptr = out_ids.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
        added = _ffi._lib.chassis_add_batch(
            self._ptr, vectors_ptr, count, self._dimensions, out_ptr
        )
        if added != count:
            raise _ffi.last_error("The batch couldn't be added")
        return out_ids

    ids = [_check_id(i) for i in (ids.tolist() if isinstance(ids, np.ndarray) else ids)]
    if len(ids) != count:
        raise InvalidArgumentError(
            f"{len(ids)} ids were given for {count} vectors\nhelp: give one id per vector, "
            "or leave ids out to have them chosen"
        )
    id_array = np.array(ids, dtype=np.uint64)
    ids_ptr = id_array.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
    if (
        _ffi._lib.chassis_add_batch_with_ids(
            self._ptr, ids_ptr, vectors_ptr, count, self._dimensions
        )
        != 0
    ):
        raise _ffi.last_error("The batch couldn't be added")
    return id_array

delete(id)

Delete the vector with this id.

Search stops returning it immediately. The delete is durable after the next flush(); a crash before then rolls it back.

Returns:

Type	Description
bool	True if it was deleted, False if no vector has this id

Raises:

Type	Description
ChassisError	If index is closed or the delete fails
ValueError	If id is out of range

Thread Safety

Safe from any thread; writes run one at a time.

Source code in chassisdb/index.py

def delete(self, id: int) -> bool:
    """Delete the vector with this id.

    Search stops returning it immediately. The delete is durable after the
    next flush(); a crash before then rolls it back.

    Returns:
        True if it was deleted, False if no vector has this id

    Raises:
        ChassisError: If index is closed or the delete fails
        ValueError: If id is out of range

    Thread Safety:
        Safe from any thread; writes run one at a time.
    """
    self._check_writable("delete")
    result: int = _ffi._lib.chassis_delete(self._ptr, _check_id(id))
    if result < 0:
        raise _ffi.last_error("The vector couldn't be deleted")
    return result == 1

get(id)

The vector with this id, or None if no live vector has it.

It comes back as the index keeps it: scaled to length 1 in a cosine index, and rounded in half precision.

Raises:

Type	Description
ChassisError	If index is closed or the file can't be read
ValueError	If id is out of range

Thread Safety

Safe from any thread. A read-only index sees the writer's last flush.

Source code in chassisdb/index.py

def get(self, id: int) -> npt.NDArray[np.float32] | None:
    """The vector with this id, or None if no live vector has it.

    It comes back as the index keeps it: scaled to length 1 in a cosine
    index, and rounded in half precision.

    Raises:
        ChassisError: If index is closed or the file can't be read
        ValueError: If id is out of range

    Thread Safety:
        Safe from any thread. A read-only index sees the writer's last
        flush.
    """
    self._check_closed()
    out = np.empty(self._dimensions, dtype=np.float32)
    out_ptr = out.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
    result: int = _ffi._lib.chassis_get(self._ptr, _check_id(id), out_ptr, self._dimensions)
    if result < 0:
        raise _ffi.last_error("The vector couldn't be read")
    return out if result == 1 else None

__contains__(id)

Whether a live vector has this id: id in index.

Source code in chassisdb/index.py

def __contains__(self, id: object) -> bool:
    """Whether a live vector has this id: ``id in index``."""
    self._check_closed()
    try:
        id = _check_id(id)
    except InvalidArgumentError:
        return False  # no vector has an id out of range
    result: int = _ffi._lib.chassis_contains(self._ptr, id)
    if result < 0:
        raise _ffi.last_error("The id couldn't be looked up")
    return result == 1

ids()

The ids of the live vectors, in the order they were added.

Thread Safety

Safe from any thread. A read-only index sees the writer's last flush.

Source code in chassisdb/index.py

def ids(self) -> npt.NDArray[np.uint64]:
    """The ids of the live vectors, in the order they were added.

    Thread Safety:
        Safe from any thread. A read-only index sees the writer's last
        flush.
    """
    self._check_closed()
    out = np.empty(len(self), dtype=np.uint64)
    while True:
        out_ptr = out.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
        count = _ffi._lib.chassis_ids(self._ptr, out_ptr, len(out))
        if count == 0 and _ffi._lib.chassis_last_error_code() != 0:
            raise _ffi.last_error("The ids couldn't be read")
        if count <= len(out):
            return out[:count]
        # A writer in another process may flush more before the next call.
        out = np.empty(count, dtype=np.uint64)

search(query, k=10, allowed=None, excluded=None)

Search for k nearest neighbors.

Parameters:

Name	Type	Description	Default
query	Sequence[float] | NDArray[float32]	Query vector (must match index dimensions)	required
k	int	Number of nearest neighbors to return (default: 10)	10
allowed	Iterable[int] | NDArray[uint64] | None	If given, only these ids can be returned. When walking the graph would cost more, as when few vectors match, every vector is checked instead and the results are exact. A numpy array of ids in ascending order costs the least; don't pass every id just to keep stray vectors out.	None
excluded	Iterable[int] | NDArray[uint64] | None	If given, these ids are never returned, as for "more like this" leaving out the vector it starts from.	None

Returns:

Type	Description
list[SearchResult]	List of SearchResult objects, sorted by distance (ascending)

Raises:

Type	Description
ChassisError	If index is closed
DimensionMismatchError	If query dimensions don't match
ValueError	If k < 1
ChassisError	For other errors

Thread Safety

Safe from any thread. Runs concurrently with other searches and waits while a write runs.

Source code in chassisdb/index.py

def search(
    self,
    query: Sequence[float] | npt.NDArray[np.float32],
    k: int = 10,
    allowed: Iterable[int] | npt.NDArray[np.uint64] | None = None,
    excluded: Iterable[int] | npt.NDArray[np.uint64] | None = None,
) -> list[SearchResult]:
    """Search for k nearest neighbors.

    Args:
        query: Query vector (must match index dimensions)
        k: Number of nearest neighbors to return (default: 10)
        allowed: If given, only these ids can be returned. When walking
            the graph would cost more, as when few vectors match, every
            vector is checked instead and the results are exact. A
            numpy array of ids in ascending order costs the least; don't
            pass every id just to keep stray vectors out.
        excluded: If given, these ids are never returned, as for "more
            like this" leaving out the vector it starts from.

    Returns:
        List of SearchResult objects, sorted by distance (ascending)

    Raises:
        ChassisError: If index is closed
        DimensionMismatchError: If query dimensions don't match
        ValueError: If k < 1
        ChassisError: For other errors

    Thread Safety:
        Safe from any thread. Runs concurrently with other searches and
        waits while a write runs.
    """
    self._check_closed()
    k = _check_int("k", k)
    if k < 1:
        raise InvalidArgumentError(
            f"k is {k}, but has to be at least 1\nhelp: pass how many results to return, "
            "such as k=10"
        )
    query = _as_vector(query, self._dimensions, "query")

    # Allocate output buffers
    out_ids = np.zeros(k, dtype=np.uint64)
    out_dists = np.zeros(k, dtype=np.float32)

    # Call FFI
    query_ptr = query.ctypes.data_as(ctypes.POINTER(ctypes.c_float))
    ids_ptr = out_ids.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
    dists_ptr = out_dists.ctypes.data_as(ctypes.POINTER(ctypes.c_float))

    if excluded is not None:
        excluded = _id_array("excluded", excluded)
    if allowed is not None:
        allowed = _id_array("allowed", allowed)
        if excluded is not None:
            allowed = np.setdiff1d(allowed, excluded)

    if allowed is not None:
        allowed_ptr = allowed.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
        count = _ffi._lib.chassis_search_filtered(
            self._ptr, query_ptr, len(query), k, allowed_ptr, len(allowed), ids_ptr, dists_ptr
        )
    elif excluded is not None:
        excluded_ptr = excluded.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64))
        count = _ffi._lib.chassis_search_excluding(
            self._ptr, query_ptr, len(query), k, excluded_ptr, len(excluded), ids_ptr, dists_ptr
        )
    else:
        count = _ffi._lib.chassis_search(
            self._ptr, query_ptr, len(query), k, ids_ptr, dists_ptr
        )

    # 0 results is an error only if one was set: an empty index has none either.
    if count == 0 and _ffi._lib.chassis_last_error_code() != 0:
        raise _ffi.last_error("The search failed")

    # Convert to SearchResult objects
    results = [
        SearchResult(id=int(out_ids[i]), distance=float(out_dists[i])) for i in range(count)
    ]

    return results

search_batch(queries, k=10)

Search for the k nearest neighbors of many queries, in one call.

Faster than calling search once per query: the queries cross from Python into the library together.

Parameters:

Name	Type	Description	Default
queries	Sequence[Sequence[float]] | NDArray[float32]	A (count, dimensions) array, one query per row	required
k	int	Number of nearest neighbors per query (default: 10)	10

Returns:

Type	Description
NDArray[uint64]	(ids, distances), two (count, k) arrays. Row i holds query i's
NDArray[float32]	results, nearest first; a row with fewer than k results is filled
tuple[NDArray[uint64], NDArray[float32]]	out with id 2**64 - 1, which no vector has, and distance inf.

Raises:

Type	Description
ChassisError	If index is closed
DimensionMismatchError	If the queries aren't rows of the index's dimensions; the message names a query that is wrong
InvalidArgumentError	If k < 1, or a query has a component that isn't a finite number
ChassisError	For other errors

Thread Safety

Safe from any thread. Runs concurrently with other searches, and a write waits for the whole batch.

Source code in chassisdb/index.py

def search_batch(
    self,
    queries: Sequence[Sequence[float]] | npt.NDArray[np.float32],
    k: int = 10,
) -> tuple[npt.NDArray[np.uint64], npt.NDArray[np.float32]]:
    """Search for the k nearest neighbors of many queries, in one call.

    Faster than calling search once per query: the queries cross from
    Python into the library together.

    Args:
        queries: A (count, dimensions) array, one query per row
        k: Number of nearest neighbors per query (default: 10)

    Returns:
        (ids, distances), two (count, k) arrays. Row i holds query i's
        results, nearest first; a row with fewer than k results is filled
        out with id 2**64 - 1, which no vector has, and distance inf.

    Raises:
        ChassisError: If index is closed
        DimensionMismatchError: If the queries aren't rows of the index's
            dimensions; the message names a query that is wrong
        InvalidArgumentError: If k < 1, or a query has a component that
            isn't a finite number
        ChassisError: For other errors

    Thread Safety:
        Safe from any thread. Runs concurrently with other searches, and a
        write waits for the whole batch.
    """
    self._check_closed()
    k = _check_int("k", k)
    if k < 1:
        raise InvalidArgumentError(
            f"k is {k}, but has to be at least 1\nhelp: pass how many results to return per "
            "query, such as k=10"
        )
    queries = _as_batch(queries, self._dimensions, "query", "search")
    count = len(queries)
    out_ids = np.empty((count, k), dtype=np.uint64)
    out_dists = np.empty((count, k), dtype=np.float32)
    if count == 0:
        return out_ids, out_dists

    done = _ffi._lib.chassis_search_batch(
        self._ptr,
        queries.ctypes.data_as(ctypes.POINTER(ctypes.c_float)),
        count,
        self._dimensions,
        k,
        out_ids.ctypes.data_as(ctypes.POINTER(ctypes.c_uint64)),
        out_dists.ctypes.data_as(ctypes.POINTER(ctypes.c_float)),
    )
    if done != count:
        raise _ffi.last_error("The search failed")
    return out_ids, out_dists

flush()

Flush all changes to disk.

This method ensures durability by writing all pending changes to disk. It's expensive (1-50ms) so batch multiple add() calls before flushing.

Raises:

Type	Description
ChassisError	If flush fails

Thread Safety

Safe from any thread; writes run one at a time.

Source code in chassisdb/index.py

def flush(self) -> None:
    """Flush all changes to disk.

    This method ensures durability by writing all pending changes to disk.
    It's expensive (1-50ms) so batch multiple add() calls before flushing.

    Raises:
        ChassisError: If flush fails

    Thread Safety:
        Safe from any thread; writes run one at a time.
    """
    self._check_writable("flush")
    if _ffi._lib.chassis_flush(self._ptr) != 0:
        raise _ffi.last_error("The flush failed")

compact()

Rewrite the index without its deleted vectors, with a newly built graph.

Reclaims the space of deleted vectors and replaces the index file with the copy. Ids don't change. Like flush(), it makes every add and delete so far durable. Takes as long as building the index, on every core, and needs free disk for a second copy of the live vectors. Readers in other processes keep searching and move to the new file by themselves.

Raises:

Type	Description
ChassisError	If the copy can't be written or, on Windows, another process has the index open. The index is left as it was.

Source code in chassisdb/index.py

def compact(self) -> None:
    """Rewrite the index without its deleted vectors, with a newly built graph.

    Reclaims the space of deleted vectors and replaces the index file with
    the copy. Ids don't change. Like flush(), it makes every add and
    delete so far durable. Takes as long as building the index, on every
    core, and needs free disk for a second copy of the live vectors.
    Readers in other processes keep searching and move to the new file by
    themselves.

    Raises:
        ChassisError: If the copy can't be written or, on Windows, another
            process has the index open. The index is left as it was.
    """
    self._check_writable("compact")
    if _ffi._lib.chassis_compact(self._ptr) != 0:
        raise _ffi.last_error("The compaction failed")

warm()

Start reading the whole index into memory on another thread.

Doesn't wait for the reading: searches go on meanwhile. While it is under way, calling again does nothing. Opening with IndexOptions(warm=True) does the same from the start.

Raises:

Type	Description
ChassisError	If the index can't be used any more

Source code in chassisdb/index.py

def warm(self) -> None:
    """Start reading the whole index into memory on another thread.

    Doesn't wait for the reading: searches go on meanwhile. While it
    is under way, calling again does nothing. Opening with
    ``IndexOptions(warm=True)`` does the same from the start.

    Raises:
        ChassisError: If the index can't be used any more
    """
    self._check_closed()

    if _ffi._lib.chassis_warm(self._ptr) != 0:
        raise _ffi.last_error("The index couldn't be read in")

close()

Close the index and free resources.

Adds and deletes since the last flush() are discarded, as after a crash. Leaving a with block flushes first, unless an exception is leaving it. Garbage collection closes without flushing. It's safe to call multiple times.

Source code in chassisdb/index.py

def close(self) -> None:
    """Close the index and free resources.

    Adds and deletes since the last flush() are discarded, as after a
    crash. Leaving a with block flushes first, unless an exception is
    leaving it. Garbage collection closes without flushing. It's safe to
    call multiple times.
    """
    if not self._closed and self._ptr:
        _ffi._lib.chassis_free(self._ptr)
        self._ptr = None
        self._closed = True

__len__()

Get the number of vectors in the index.

Returns:

Type	Description
int	Number of vectors

Thread Safety

Safe from any thread.

Source code in chassisdb/index.py

def __len__(self) -> int:
    """Get the number of vectors in the index.

    Returns:
        Number of vectors

    Thread Safety:
        Safe from any thread.
    """
    self._check_closed()
    return int(_ffi._lib.chassis_len(self._ptr))

deleted_count()

Deleted vectors the file still holds, until compact() removes them.

Thread Safety

Safe from any thread.

Source code in chassisdb/index.py

def deleted_count(self) -> int:
    """Deleted vectors the file still holds, until compact() removes them.

    Thread Safety:
        Safe from any thread.
    """
    self._check_closed()
    return int(_ffi._lib.chassis_deleted_count(self._ptr))

is_empty()

Check if the index is empty.

Returns:

Type	Description
bool	True if empty, False otherwise

Thread Safety

Safe from any thread.

Source code in chassisdb/index.py

def is_empty(self) -> bool:
    """Check if the index is empty.

    Returns:
        True if empty, False otherwise

    Thread Safety:
        Safe from any thread.
    """
    self._check_closed()
    return bool(_ffi._lib.chassis_is_empty(self._ptr))