in src/dfcx_scrapi/core/search.py [0:0]
def search(self, search_config: Dict[str, Any], total_results: int = 10):
"""Performs a search against an indexed Vertex Data Store.
Args:
search_config: A dictionary containing keys that correspond to the
SearchRequest attributes as defined in: https://cloud.google.com/python/docs/reference/discoveryengine/latest/google.cloud.discoveryengine.SearchRequest
For complex attributes that require nested fields, you can pass
in another Dictionary as the value.
Example: To represent the complex facet_specs config with some
other simple parameters, you would do the following.
```py
search_config = {
"facet_specs": [
{
"facet_key": {
"key": "my_key",
"intervals": [
{
"minimum": .5
},
{
"maximum": .95
}
],
"case_insensitive": True
},
"limit": 10
}
],
"page_size": 10,
"offset": 2
}
total_results: Total number of results to return for the search. If
not specified, will default to 10 results. Increasing this to a
high number can result in long search times.
Returns:
A List of SearchResponse objects.
"""
serving_config = (
f"{search_config.get('data_store_id', None)}"
"/servingConfigs/default_serving_config"
)
branch_stub = "/".join(serving_config.split("/")[0:8])
branch = branch_stub + "/branches/0"
request = SearchRequest(
serving_config=serving_config,
branch=branch,
query=search_config.get("query", None),
image_query=self.build_image_query(search_config),
page_size=search_config.get("page_size", 10),
page_token=search_config.get("page_token", None),
offset=search_config.get("offset", 0),
filter=search_config.get("filter", None),
canonical_filter=search_config.get("canonical_filter", None),
order_by=search_config.get("order_by", None),
user_info=self.build_user_info(search_config),
facet_specs=self.build_facet_specs(search_config),
boost_spec=self.build_boost_spec(search_config),
params=search_config.get("params", None),
query_expansion_spec=self.build_query_expansion_spec(search_config),
spell_correction_spec=self.build_spell_correction_spec(
search_config
),
user_pseudo_id=search_config.get("user_pseudo_id", None),
content_search_spec=self.build_content_search_spec(search_config),
embedding_spec=self.build_embedding_spec(search_config),
ranking_expression=search_config.get("ranking_expression", None),
safe_search=search_config.get("safe_search", False),
user_labels=search_config.get("user_labels", None),
)
client_options = self._client_options_discovery_engine(serving_config)
client = SearchServiceClient(
credentials=self.creds, client_options=client_options
)
response = client.search(request)
all_results = []
for search_result in response:
if len(all_results) < total_results:
all_results.append(search_result)
else:
break
return all_results