"""
Implementation of full text search using Elasticsearch or Meilisearch.
The Elasticsearch API is still for python 7.17. There seems to be some change for version 8.0 on.
"""
import typing
from flask import current_app
[docs]def add_to_index(index: str, model) -> None:
"""
Add entries to search index.
Add the fields marked in __searchable__ to the elasticsearch index.
Parameters
----------
index: str
Index name
model: db.Model
Returns
-------
None
"""
if not current_app.elasticsearch and not current_app.meilisearch:
return
if current_app.elasticsearch:
payload = {}
for field in model.__searchable__:
payload[field] = getattr(model, field)
current_app.logger.debug(f"Adding {model.id} to index {index}")
current_app.elasticsearch.index(index=index, id=model.id, document=payload)
if current_app.meilisearch:
m_index = current_app.meilisearch.index(index)
payload = {}
payload["id"] = model.id
for field in model.__searchable__:
payload[field] = getattr(model, field)
current_app.logger.debug(f"Adding {payload} to index {index}")
m_index.add_documents([payload])
[docs]def remove_from_index(index: str, model) -> None:
"""
Remove entry from index.
This seems not to work in our test case
Parameters
----------
index : str
Index name
model : db.Model
Returns
-------
None
"""
if not current_app.elasticsearch and not current_app.meilisearch:
return
if current_app.elasticsearch:
current_app.logger.debug(f"Removing {model.id} from index")
current_app.elasticsearch.delete(index=index, id=model.id)
if current_app.meilisearch:
m_index = current_app.meilisearch.index(index)
current_app.logger.debug(f"Removing {model.id} from index")
m_index.delete_document(model.id)
[docs]def delete_index(index: str) -> None:
"""
Delete search index.
Parameters
----------
index: str
Index name.
Returns
-------
None
"""
if not current_app.elasticsearch and not current_app.meilisearch:
return
if current_app.elasticsearch:
current_app.elasticsearch.indices.delete(index=index, ignore=[400, 404])
if current_app.meilisearch:
current_app.meilisearch.index(index).delete()
return
[docs]def create_index(index: str) -> None:
"""
Create the search index.
Parameters
----------
index: str
UID of the index to be created.
Returns
-------
"""
if current_app.meilisearch:
current_app.meilisearch.create_index(index)
return
[docs]def query_index(index: str, query: str) -> typing.Tuple[typing.List[int], int]:
"""
Search string in index.
Searches the query term inside the index. We get a despondency warning and have to redo the command.
We have to read the search API of the elasticsearch module. We will have problems from version 8 on.
Parameters
----------
index: str
Index name
query: str
Search term (we might want to rename this).
Returns
-------
ids: List[int]
Found document ids.
number of hits : int
Number of found entries.
"""
if not current_app.elasticsearch and not current_app.meilisearch:
return [], 0
if current_app.elasticsearch:
search = current_app.elasticsearch.search(
index=index,
body={
"query": {"multi_match": {"query": query, "fields": ["*"], "fuzziness": "AUTO"}},
},
)
current_app.logger.debug(f'Found {search["hits"]["total"]["value"]} searching index {index}')
ids = [int(hit["_id"]) for hit in search["hits"]["hits"]]
return ids, search["hits"]["total"]["value"]
if current_app.meilisearch:
m_index = current_app.meilisearch.index(index)
search = m_index.search(query)
ids = [int(h["id"]) for h in search["hits"]]
current_app.logger.debug(f"Found searching index {index} {len(search['hits'])} hits")
current_app.logger.debug(f"List of IDs: {ids}")
return ids, len(ids)
return [], 0