1111# and limitations under the License.
1212#
1313
14+ import logging
15+ import os
16+ import time
17+ from typing import Sequence , TypedDict
18+
19+ import boto3
20+ from custom_resources .cr_types import CustomResourceRequest , CustomResourceResponse
1421from opensearchpy import (
22+ AuthorizationException ,
23+ AWSV4SignerAuth ,
1524 OpenSearch ,
1625 RequestsHttpConnection ,
17- AWSV4SignerAuth ,
18- AuthorizationException ,
1926)
20- import boto3
21- import logging
22- import os
23- import time
2427from tenacity import (
2528 retry ,
2629 retry_if_exception_type ,
2730 stop_after_attempt ,
2831 wait_exponential_jitter ,
2932)
3033
31- from typing import TypedDict , Sequence
32-
33- from custom_resources .cr_types import CustomResourceRequest , CustomResourceResponse
34-
3534LOG_LEVEL = os .getenv ("LOG_LEVEL" , "INFO" )
3635
3736logger = logging .getLogger (__name__ )
@@ -44,12 +43,19 @@ class MetadataManagementField(TypedDict):
4443 Filterable : bool
4544
4645
46+ class AnalyzerProperties (TypedDict ):
47+ CharacterFilters : Sequence [str ]
48+ Tokenizer : str
49+ TokenFilters : Sequence [str ]
50+
51+
4752class VectorIndexProperties (TypedDict ):
4853 Endpoint : str
4954 IndexName : str
5055 VectorField : str
5156 Dimensions : int | str
5257 MetadataManagement : Sequence [MetadataManagementField ]
58+ Analyzer : AnalyzerProperties | None
5359
5460
5561def validate_event (event : CustomResourceRequest [VectorIndexProperties ]) -> bool :
@@ -70,6 +76,14 @@ def validate_event(event: CustomResourceRequest[VectorIndexProperties]) -> bool:
7076 raise ValueError ("MetadataManagement is required" )
7177 if event ["RequestType" ] == "Update" and event ["PhysicalResourceId" ] is None :
7278 raise ValueError ("PhysicalResourceId is required" )
79+ if event ["ResourceProperties" ].get ("Analyzer" ) is not None :
80+ analyzer = event ["ResourceProperties" ]["Analyzer" ]
81+ if analyzer ["CharacterFilters" ] is None :
82+ raise ValueError ("CharacterFilters is required" )
83+ if analyzer ["Tokenizer" ] is None :
84+ raise ValueError ("Tokenizer is required" )
85+ if analyzer ["TokenFilters" ] is None :
86+ raise ValueError ("TokenFilters is required" )
7387 elif event ["RequestType" ] == "Delete" :
7488 if event ["PhysicalResourceId" ] is None :
7589 raise ValueError ("PhysicalResourceId is required" )
@@ -139,18 +153,39 @@ def create_mapping(
139153 return mapping
140154
141155
142- def create_index (client : OpenSearch , index_name : str , mapping : dict [str , str ]) -> None :
156+ def create_setting (analyzer : AnalyzerProperties | None ) -> dict :
157+ setting = {
158+ "index" : {
159+ "number_of_shards" : "2" ,
160+ "knn.algo_param" : {"ef_search" : "512" },
161+ "knn" : "true" ,
162+ },
163+ }
164+ if analyzer :
165+ setting ["analysis" ] = {
166+ "analyzer" : {
167+ "custom_analyzer" : {
168+ "type" : "custom" ,
169+ "tokenizer" : analyzer ["Tokenizer" ],
170+ "char_filter" : analyzer ["CharacterFilters" ],
171+ "filter" : analyzer ["TokenFilters" ],
172+ }
173+ }
174+ }
175+
176+ return setting
177+
178+
179+ def create_index (
180+ client : OpenSearch , index_name : str , mapping : dict [str , str ], setting : dict [str , str ]
181+ ) -> None :
143182 logger .debug (f"creating index { index_name } )
183+ logger .debug (f"setting: { setting } )
184+ logger .debug (f"mapping: { mapping } )
144185 client .indices .create (
145186 index_name ,
146187 body = {
147- "settings" : {
148- "index" : {
149- "number_of_shards" : "2" ,
150- "knn.algo_param" : {"ef_search" : "512" },
151- "knn" : "true" ,
152- }
153- },
188+ "settings" : setting ,
154189 "mappings" : mapping ,
155190 },
156191 params = {"wait_for_active_shards" : "all" },
@@ -171,13 +206,15 @@ def handle_create(
171206 vector_field : str ,
172207 dimensions : int ,
173208 metadata_management : Sequence [MetadataManagementField ],
209+ analyzer : AnalyzerProperties | None ,
174210):
175211 if client .indices .exists (index_name ):
176212 raise ValueError (f"Index { index_name } )
177213
178214 try :
179215 mapping = create_mapping (vector_field , dimensions , metadata_management )
180- create_index (client , index_name , mapping )
216+ setting = create_setting (analyzer )
217+ create_index (client , index_name , mapping , setting )
181218 except Exception as e :
182219 logger .error (f"Error creating index { index_name } )
183220 logger .exception (e )
@@ -211,6 +248,7 @@ def on_create(
211248 event ["ResourceProperties" ]["VectorField" ],
212249 int (event ["ResourceProperties" ]["Dimensions" ]),
213250 event ["ResourceProperties" ]["MetadataManagement" ],
251+ event ["ResourceProperties" ].get ("Analyzer" , None ),
214252 )
215253 return {"PhysicalResourceId" : physical_id }
216254
0 commit comments