"""
Define all mixin classes for later reimport
"""
# from dataclasses import dataclass
# from datetime import datetime
from typing import Any # Needs to be here for compatibility with python 3.8
from ResearchNotes.search import add_to_index, remove_from_index, query_index, delete_index, create_index
from ResearchNotes.database import db
# =============================================================================
# Define our Searchable Mixin class
# =============================================================================
[docs]class SearchableMixin:
"""
Mixin class for database to add search function in fulltext search engine.
We define some functions to index automatically some fields in a search engine database. Also, a
search function is provided to retrieve information via a search query.
This works currently with Elasticsearch and Meilisearch.
"""
__tablename__: str
id: int
[docs] @classmethod
def search(cls, expression: str) -> tuple[Any, int]:
"""
Search expression inside the index of a fulltext search engine.
Parameters
----------
expression : str
String to be searched in the index. We call the query_index function as defined in the
search submodule.
Returns
-------
tuple[SQLAlchemy database record, Int]
"""
ids, total = query_index(cls.__tablename__, expression)
if total == 0:
return db.session.execute(db.select(cls).filter_by(id=0)).scalars(), 0
# return cls.query.filter(cls.id.in_(ids)).order_by(db.case(when, value=cls.id)), total
# print(ids)
# result = (db.session.execute(db.select(cls).where(cls.id.in_(ids))).scalars(), total)
result = ([db.session.get(cls, oid) for oid in ids], total)
return result
[docs] @classmethod
def before_commit(cls, session) -> None:
"""
Remember objects of database session as changed before the commit to the database.
We need to remember our objects that changed (like new, updated or deleted) and add, update or
delete them accordingly form our search index. Hence, the function is called before every commit of
a session by SQLAlachemy (see below as registered as event.listen to db object).
Parameters
----------
session : SQLAlchemy class or object
session of our database (should be db.session most of the time).
Returns
-------
None
"""
session._changes = {
"add": list(session.new),
"update": list(session.dirty),
"delete": list(session.deleted),
}
[docs] @classmethod
def after_commit(cls, session) -> None:
"""
Function to be called after the commit to the database.
After successful commit to the database, this function is called to add, update or delete entries
in our search index.
Parameters
----------
session : SQLAlchemy.session
Session for our database
Returns
-------
None
"""
for obj in session._changes["add"]:
if isinstance(obj, SearchableMixin):
add_to_index(obj.__tablename__, obj)
for obj in session._changes["update"]:
if isinstance(obj, SearchableMixin):
add_to_index(obj.__tablename__, obj)
for obj in session._changes["delete"]:
if isinstance(obj, SearchableMixin):
remove_from_index(obj.__tablename__, obj)
session._changes = None
[docs] @classmethod
def reindex(cls):
"""
Reindex a database model in the fulltext search engine.
This will delete the index of the database model in the full text search engine database and
rewrite the index for the db.Model adding all fields marked as searchable.
Needs to be run the first time or if the search index somehow lost the connection to the
actual database entries, e.g. if the search engine was down and new entries have been committed.
Returns
-------
None
"""
delete_index(cls.__tablename__)
create_index(cls.__tablename__)
for obj in db.session.execute(db.select(cls)).scalars():
add_to_index(cls.__tablename__, obj)
[docs] @classmethod
def make_index(cls) -> None:
"""
Create the search index for the class.
Returns
-------
"""
create_index(cls.__tablename__)
# Registers the two before and after commit functions to the db.session. We take write dirctly to the
# event lister of the session as explained in the SQLAlchemy documentation (which is bad).
db.event.listen(db.session, "before_commit", SearchableMixin.before_commit)
db.event.listen(db.session, "after_commit", SearchableMixin.after_commit)