oasysdb 0.6.0

Fast embedded vector database with incremental HNSW indexing.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

import asyncio
from oasysdb.prelude import Record, Collection, Config, Database


NAME = "vectors"  # Initial collection name.
DIMENSION = 128
LEN = 100


def create_test_database() -> Database:
    """Creates a new test database with an initial collection."""

    db = Database.new("data/py")
    assert db.is_empty()

    # Create a test collection with random records.
    records = Record.many_random(dimension=DIMENSION, len=LEN)
    config = Config.create_default()
    collection = Collection.from_records(config, records)

    # Save the collection to the database.
    db.save_collection(name=NAME, collection=collection)
    assert not db.is_empty()

    return db


def test_open():
    db = Database(path="data/mt")
    assert db.is_empty()


def test_new():
    db = create_test_database()
    assert not db.is_empty()
    assert db.len() == 1


def test_get_collection():
    db = create_test_database()
    collection = db.get_collection(name=NAME)
    assert collection.len() == LEN


def test_save_collection():
    db = create_test_database()

    # Create a new collection and save it to the database.
    config = Config.create_default()
    collection = Collection(config=config)
    db.save_collection(name="test", collection=collection)

    assert db.len() == 2


def test_delete_collection():
    db = create_test_database()
    db.delete_collection(name=NAME)
    assert db.is_empty()


def test_flush():
    db = create_test_database()
    assert db.flush() > 0


def test_async_flush():
    db = create_test_database()
    assert asyncio.run(db.async_flush()) > 0