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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
import json
from typing import Any, Optional
from shared.api.models import (
WrappedGenericMessageResponse,
WrappedVectorIndexResponse,
WrappedVectorIndicesResponse,
)
class IndicesSDK:
def __init__(self, client):
self.client = client
def create(
self,
config: dict,
run_with_orchestration: Optional[bool] = True,
) -> WrappedGenericMessageResponse:
"""Create a new vector similarity search index in the database.
Args:
config (dict | IndexConfig): Configuration for the vector index.
run_with_orchestration (Optional[bool]): Whether to run index creation as an orchestrated task.
Returns:
WrappedGenericMessageResponse
"""
if not isinstance(config, dict):
config = config.model_dump()
data: dict[str, Any] = {
"config": config,
"run_with_orchestration": run_with_orchestration,
}
response_dict = self.client._make_request(
"POST",
"indices",
json=data,
version="v3",
)
return WrappedGenericMessageResponse(**response_dict)
def list(
self,
filters: Optional[dict] = None,
offset: Optional[int] = 0,
limit: Optional[int] = 10,
) -> WrappedVectorIndicesResponse:
"""List existing vector similarity search indices with pagination
support.
Args:
filters (Optional[dict]): Filter criteria for indices.
offset (int, optional): Specifies the number of objects to skip. Defaults to 0.
limit (int, optional): Specifies a limit on the number of objects to return, ranging between 1 and 100. Defaults to 100.
Returns:
WrappedVectorIndicesResponse
"""
params: dict = {
"offset": offset,
"limit": limit,
}
if filters:
params["filters"] = json.dumps(filters)
response_dict = self.client._make_request(
"GET",
"indices",
params=params,
version="v3",
)
return WrappedVectorIndicesResponse(**response_dict)
def retrieve(
self,
index_name: str,
table_name: str = "vectors",
) -> WrappedVectorIndexResponse:
"""Get detailed information about a specific vector index.
Args:
index_name (str): The name of the index to retrieve.
table_name (str): The name of the table where the index is stored.
Returns:
WrappedGetIndexResponse
"""
response_dict = self.client._make_request(
"GET",
f"indices/{table_name}/{index_name}",
version="v3",
)
return WrappedVectorIndexResponse(**response_dict)
def delete(
self,
index_name: str,
table_name: str = "vectors",
) -> WrappedGenericMessageResponse:
"""Delete an existing vector index.
Args:
index_name (str): The name of the index to retrieve.
table_name (str): The name of the table where the index is stored.
Returns:
WrappedGetIndexResponse
"""
response_dict = self.client._make_request(
"DELETE",
f"indices/{table_name}/{index_name}",
version="v3",
)
return WrappedGenericMessageResponse(**response_dict)
|
